AbstractQueuedSynchronizer之AQS

时间：2021-03-12 15:19:52 阅读：15 评论：0 收藏：0 [点我收藏+]

一、可重入锁

1，定义

指的是同一线程外层函数获得锁后，再进入该线程的内层方法会自动获取锁(前提：锁对象是同一个对象)。
Java中的ReentranLock(显示锁)和Synchronized(隐式锁)都是可重入锁，可重入锁的一个优点是可在一定程度避免死锁
隐士锁：(即synchronized关键字使用的锁)默认是可重入锁(同步块、同步方法)

2，案例

　　synchronize隐式锁

public class Demo01_ReentrantLockSynchronizedMethod {

    public static void main(String[] args) {
        new Demo01_ReentrantLockSynchronizedMethod().m1();
    }

    private synchronized void m1() {
        System.out.println("=====外层");
        m2();
    }

    private synchronized void m2() {
        System.out.println("=====中层");
        m3();
    }

    private synchronized void m3() {
        System.out.println("=====内层");

    }

}

View Code

　　ReentrantLock显示锁

public class Demo01_ReentrantLockShow {

    static Lock lock = new ReentrantLock();

    public static void main(String[] args) {
        new Thread(()->{
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + "==============外部");
                lock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + "==============内部");
                }finally {
                    lock.unlock();
                }
            }finally {
                lock.unlock();
//                lock.unlock();
            }

        },"t1").start();

        new Thread(()->{
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + "==========进入方法");
            }finally {
                lock.unlock();
            }
        },"t2").start();
    }

}

View Code

3，原理

每个锁对象拥有一个锁计数器和一个指向持有该锁的线程的指针
当执行monitorenter时,如果目标锁对象的计数器为零,那么说明它没有被其他线程持有,Java虚拟机会将该锁对象的持有线程设置为当前线程,并且将其计数器加1,否则需要等待,直至持有线程释放该锁
当执行monitorexit时,Java虚拟机则锁对象的计数器减1。计数器为零代表锁已经被释放

二、LockSupport

1，三种线程唤醒等待

a）synchronized，Object的wait和notify

　　代码

private static void SynchroziedWaitNotify() {
    new Thread(() -> {
        //如果注释掉，就会先执行进入程序等待被唤醒
        try { Thread.sleep(3000); } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //如果注释掉synchronized 则会报错，因为wait和notify一定要在同步块或同步方法中
        synchronized (objectLock) {
            try {
                System.out.println(Thread.currentThread().getName() + "=========进入");
                objectLock.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "=========结束");
        }
    }, "t1").start();


    new Thread(() -> {
        synchronized (objectLock) {
            objectLock.notify();
            System.out.println(Thread.currentThread().getName() + "=========唤醒");
        }
    }, "t2").start();
}

　　wait和notify的限制条件：

wait和notify方法必须要在同步块或同步方法里且成对出现使用。
先wait后notify才可以(如果先notify后wait会出现另一个线程一直处于等待状态)
synchronized是关键字属于JVM层面。monitorenter(底层是通过monitor对象来完成,其实wait/notify等方法也依赖monitor对象只能在同步块或方法中才能调用wait/notify等方法)

b）Lock，Condition的await和signal

private static void LockAwaitSignal() {
    new Thread(() -> {
        //如果把下行这句代码打开,先signal后await,会出现A线程一直处于等待状态
        try { Thread.sleep(3000); } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //如果不加lock锁也会出现错误同synchronize
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + "=========进入");
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        System.out.println(Thread.currentThread().getName() + "=========结束");
    }, "t1").start();


    new Thread(() -> {
        lock.lock();
        try {
            condition.signal();
            System.out.println(Thread.currentThread().getName() + "=========唤醒");
        }finally {
            lock.unlock();
        }
    }, "t2").start();
}

　　await和signal的限制条件：

await和signal都需要许出现在lock中，否则会报错
必须先await再signal否则会出现线程等待

c）LockSupport的park和unpark

　　代码

private static void lockSupportParkUnpark() {
    Thread t1 = new Thread(() -> {
        try { Thread.sleep(3000); } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "=========进入");
        //如果这里有两个LockSupport.park(),因为permit的值为1,上一行已经使用了permit，
        // 所以下一行被注释的打开会导致程序处于一直等待的状态
        LockSupport.park();
        LockSupport.park();
        System.out.println(Thread.currentThread().getName() + "=========结束");
    }, "t1");
    t1.start();


    new Thread(() -> {
        //有两个LockSupport.unpark(t1),由于permit的值最大为1,所以只能给park一个通行证
        LockSupport.unpark(t1);
        LockSupport.unpark(t1);
        System.out.println(Thread.currentThread().getName() + "=========唤醒");
    }, "t2").start();
}

2，LockSupport的描述

LockSupport是用来创建锁和其他同步类的基本线程阻塞原语
LockSupport是一个线程阻塞工具类，所有的方法都是静态方法，可以让线程在任意位置阻塞，阻塞之后也有对应的唤醒方法。归根结底，LockSupport调用的Unsafe中的native代码。
LockSupport提供park()和unpark()方法实现阻塞线程和解除线程阻塞的过程LockSupport和每个使用它的线程都有一个许可(permit)关联。permit相当于1，0的开关，默认是0，每个线程都有一个相关的permit, permit最多只有一个，重复调用unpark也不会积累凭证。

3，LockSupport的面试题

a）为什么可以先唤醒线程后阻塞线程?
　　因为unpark获得了一个凭证，之后再调用park方法，就可以名正言顺的凭证消费，故不会阻塞。
b）为什么唤醒两次后阻塞两次，但最终结果还会阻塞线程?
　　因为凭证的数量最多为1，连续调用两次unpark和调用一次unpark效果一样，只会增加一个凭证;而调用两次park却需要消费两个凭证，证不够，不能放行。

三、AQS的架构

1，AQS是什么

　　是用来构建锁或者其它同步器组件的重量级基础框架及整个JUC体系的基石，通过内置的CLH(FIFO)队列的变种来完成资源获取线程的排队工作,将每条将要去抢占资源的线程封装成一个Node节点来实现锁的分配，有一个int类变量表示持有锁的状态,通过CAS完成对state值的修改(0表示没有,1表示阻塞次数用于记录可重入)

　　技术分享图片

2，AQS的内部结构体系

　　技术分享图片

四，ReentrantLock非公平锁之lock

1，NonfairSync 继承Sync

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    //非公平锁加锁
    final void lock() {
        //首先尝试修改 state 如果能从0修改为1 则表示当前还没有对象加锁成功
        if (compareAndSetState(0, 1))
            //修改此时的线程持有者为当前线程
            setExclusiveOwnerThread(Thread.currentThread());
        else
            //否则就表示当前已经有线程持有锁。此时开始尝试获得锁
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

直接尝试加锁的比较并交换，compareAndSetState(0, 1)。
如果线程状态state为0，会加锁成功并修改当前线程的持有者
如果线程状态state不为0，会加锁失败，会继续一步逻辑2

2,调用AbstractQueuedSynchronizer.acquire(int arg)

//尝试获得锁
public final void acquire(int arg) {
    //1. tryAcquire(arg) 实际上是调用NonfairSync.tryAcquire(1)。表示当前线程是否获取锁成功
    //2. addWaiter(Node.EXCLUSIVE)初始化CLH链表
    //3. acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

a）第一步tryAcquire(arg)

　　先调用NonfairSync.tryAcquire(1)实际也就是a)中的方法，最终调用Sync.nonfairTryAcquire(1)方法。

final boolean nonfairTryAcquire(int acquires) {
    //获取当前线程和当前对象锁的状态
    final Thread current = Thread.currentThread();
    int c = getState();
    //如果状态为0，表示当前锁没有被占有。 修改当前状态为1，并且修改当前持有线程。并且返回获取锁成功
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    //如果 当前线程为持有线程 。nextc指针为当前线程持有锁的数量，表示可重入锁。 并且返回获取锁成功
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    //如果当前线程没有获取到锁，就返回false。
    return false;
}

这时当前线程会继续判断当前state是否为0，如果为0则直接抢占锁compareAndSetState(0, acquires)，如果能抢占成功就直接修改对象的线程持有者。这种情况直接返回true。
如果当前线程就是正在执行的线程，就会为state属性加值，也就是可重锁了。这种情况直接返回true。
如果当前线程并没有获取到锁则会直接返回false。这时会走b)第二步

b)第二步addWaiter(Node.EXCLUSIVE)

　　实际调用AbstractQueuedSynchronizer.addWaiter(Node mode)，传入的mode为null

private Node addWaiter(Node mode) {
    //初始化node结点其中Thread为当前线程
    Node node = new Node(Thread.currentThread(), mode);
    // 定义pred为尾结点，第一次调用的情况下当前结点值为null
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //如果pred为null，则执行入CLH链表操作。返回当前链表的上一次出现的尾结点，当不存在尾结点时会构建一个虚结点(但是enq这个方法没有用返回值)
    //并且把当前传入的节点直接保存到当前链表的尾部
    enq(node);
    //返回当前加入的结点
    return node;
}

始化node结点为当前线程的结点
判断tail尾结点是否为null，如果tail尾结点不为空，则将当前初始化的node结点直接加入到双向链表的最后，并令其为尾结点返回当前结点
如果tail尾结点为null，则需要执行enq(node)，来初始化话CLH链表。具体为调用AbstractQueuedSynchronizer.enq(final Node node)，其中node为当前线程构建的结点。

private Node enq(final Node node) {
    //循环自旋
    for (;;) {
        //设置尾结点为t
        Node t = tail;
        if (t == null) { 
            //初始化头结点为一个空节点，也叫做哨兵结点(虚结点)
            if (compareAndSetHead(new Node()))
                //并设置尾结点=头结点
                tail = head;
        } else {
            //第二次进入当前循环则得到设置传入结点的前面一个结点为头结点，构建双向链表
            node.prev = t;
            //第二次由于 t = tail = head，故而会比较并交换为单签node结点，也就是设置当前双向链表的尾结点为传入的node结点 
            if (compareAndSetTail(t, node)) {
                //设置当前t结点为 虚结点，设置当前虚结点的下一个结点为当前传入的结点。并返回尾结点的上一个结点(第一个线程返回的是头结点,第二个线程返回的是第一个线程的结点)
                t.next = node;
                return t;
            }
        }
    }
}

采用CAS自旋，如果tail尾结点为null，则初始化一个哨兵结点，并设置头结点和尾结点都为当前结点
如果tail尾结点不为null，就令当前传入的node结点为尾结点，保留哨兵结点。并返回尾结点的前一个结点。这样就构建了一个双向链表

c）第三步acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

　　实际调用AbstractQueuedSynchronizer.acquireQueued(final Node node, int arg)，其中node为当前线程构建的结点，arg为1

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            //设置p结点为 当前传入结点的前一个结点,第一次为头结点.第二次为第一次的结点
            final Node p = node.predecessor();
            //如果p结点为头结点 ,就会再次去尝试获取锁
            if (p == head && tryAcquire(arg)) {
                //如果当前结点获取到锁,会设置头结点为当前结点.并设置p结点的下一个结点为null
                //这样就是为了释放头结点
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //二次调用这个方法会返回true , 然后执行parkAndCheckInterrupt会将当前线程挂起
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

采用CAS自旋，先获取当前结点的前一个结点p
如果p结点为头结点，证明当前结点为队列中下一个需要被调用的线程，首先再次尝试去获取锁
如果获取成功，则将当前结点设置为头结点，并将p结点引用置空方便回收。返回false中断当前线程获取锁的操作
如果获取失败，此时调用方法shouldParkAfterFailedAcquire，将当前线程park()挂起

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //第一次使用的时候初始化为0
    int ws = pred.waitStatus;
    //SIGNAL为-1,第二次会为true
    if (ws == Node.SIGNAL) 
        //返回true
        return true;
    if (ws > 0) {
        
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //第一次走这步,会设置pred结点中的ws为-1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

第一次初始化ws为0，然后修改pre的节点waitstatus为-1，并返回false，自旋继续调用
第二次返回true，挂起线程结束循环。

3，最终实现的效果图为

　　技术分享图片

五、ReentrantLock非公平锁之unlock

1，AbstractQueuedSynchronizer.release(int arg)

　　传入参数arg为-1

public final boolean release(int arg) {
    //尝试去释放锁
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

先尝试释放锁，如果释放成功，则判断如果头结点不为空并且头结点的waitstatus不为0，就是释放队列中的下一个线程

2，尝试释放锁Sync.tryRelease(arg)

protected final boolean tryRelease(int releases) {
    //将当前状态值减去release 
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

3，调用AbstractQueuedSynchronizer.unparkSuccessor

private void unparkSuccessor(Node node) {
    //获取头结点的ws
    int ws = node.waitStatus;
    if (ws < 0)
        //如果为-1则修改为0
        compareAndSetWaitStatus(node, ws, 0);
    //定义s为下一个结点
    Node s = node.next;
    //如果下个结点为null 或者是 下个结点的waitStatus>0
    //则从为结点往前遍历,知道碰到waitStatus<=0的结点,赋值给s
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    //给s线程一个许可证，其实就是按照队列的顺序去释放锁
    if (s != null)
        LockSupport.unpark(s.thread);
}

六、对比公平锁与非公平锁的区别

　　从源码层面来看主要有两个地方不一样：

1，非公平锁当调用lock()方法时，当前线程会首先尝试去获取锁，而非公平锁会直接调用acquire(1)方法

　　技术分享图片

2，调用acquire()方法时也不同

　　技术分享图片

　　非公平锁会再次尝试修改当前锁状态，获得锁；而公平锁会调用hasQueuedPredecessors()方法首先判断是否需要入队列，再决定是否获取当前锁。

public final boolean hasQueuedPredecessors() {
    Node t = tail;
    Node h = head;
    Node s;
    //如果队列没有初始化，也就是不存在等待队列，那么t=null,h=null，会直接返回false。那么非公平锁取反会尝试获取锁
    //如果队列已经初始化，那么t肯定不等于h（因为队列初始化之后存在哨兵结点）则 h!=t -> true，那么获取到的s为头结点的下一个结点
    //如果s结点为null，则直接返回true，说明当前结点中只存在头结点，这种情况不会出现。因为队列中起码有一个元素
    //如果s结点不为null -> false，并且s.thread!=Thread.currentThread() -> false ，说明下一个将要执行的线程为当前线程则不需要排队了，尝试获取锁
    //如果s结点不为null -> false，并且s.thread!=Thread.currentThread() -> true说明下一个结点的线程不是当前线程，返回true，需要去排队
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

AbstractQueuedSynchronizer之AQS

原文：https://www.cnblogs.com/bbgs-xc/p/14517096.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)