Java中的显示锁ReentrantLock使用与AbstractQueuedSynchronizer原理剖析

考虑一个场景，轮流打印0-100以内的技术和偶数。通过使用 synchronize 的 wait，notify机制就可以实现，核心思路如下：

使用两个线程，一个打印奇数，一个打印偶数。这两个线程会共享一个数据，数据每次自增，当打印奇数的线程发现当前要打印的数字不是奇数时，执行等待，否则打印奇数，并将数字自增1，对于打印偶数的线程也是如此

//打印奇数的线程
private static class OldRunner implements Runnable{
    private MyNumber n;

    public OldRunner(MyNumber n) {
        this.n = n;
    }

    public void run() {
        while (true){
            n.waitToOld();  //等待数据变成奇数
            System.out.println("old:" + n.getVal());
            n.increase();
            if (n.getVal()>98){
                break;
            }
        }
    }
}
//打印偶数的线程
private static class EvenRunner implements Runnable{
    private MyNumber n;

    public EvenRunner(MyNumber n) {
        this.n = n;
    }

    public void run() {
        while (true){
            n.waitToEven();            //等待数据变成偶数
            System.out.println("even:"+n.getVal());
            n.increase(); 
            if (n.getVal()>99){
                break;
            }
        }
    }
}
复制代码

共享的数据如下

private static class MyNumber{
    private int val;

    public MyNumber(int val) {
        this.val = val;
    }

    public int getVal() {
        return val;
    }
    public synchronized void increase(){
        val++;
        notify(); //数据变了，唤醒另外的线程
    }
    public synchronized void  waitToOld(){
        while ((val % 2)==0){
            try {
                System.out.println("i am "+Thread.currentThread().getName()+" ,but now is even:"+val+",so wait");
                wait(); //只要是偶数，一直等待
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public synchronized void waitToEven(){
        while ((val % 2)!=0){
            try {
                System.out.println("i am "+Thread.currentThread().getName()+"  ,but now old:"+val+",so wait");
                wait(); //只要是奇数，一直等待
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
复制代码

运行代码如下

MyNumber n = new MyNumber(0);
Thread old=new Thread(new OldRunner(n),"old-thread");
Thread even = new Thread(new EvenRunner(n),"even-thread");
old.start();
even.start();
复制代码

运行结果如下

i am old-thread ,but now is even:0,so wait
even:0
i am even-thread  ,but now old:1,so wait
old:1
i am old-thread ,but now is even:2,so wait
even:2
i am even-thread  ,but now old:3,so wait
old:3
i am old-thread ,but now is even:4,so wait
even:4
i am even-thread  ,but now old:5,so wait
old:5
i am old-thread ,but now is even:6,so wait
even:6
i am even-thread  ,but now old:7,so wait
old:7
i am old-thread ,but now is even:8,so wait
even:8
复制代码

上述方法使用的是 synchronize的 wait notify机制，同样可以使用显示锁来实现，两个打印的线程还是同一个线程，只是使用的是显示锁来控制等待事件

private static class MyNumber{
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private int val;

    public MyNumber(int val) {
        this.val = val;
    }

    public int getVal() {
        return val;
    }
    public  void increase(){
        lock.lock();
        try {
            val++;
            condition.signalAll(); //通知线程
        }finally {
            lock.unlock();
        }

    }
    public  void  waitToOld(){
        lock.lock();
        try{
            while ((val % 2)==0){
                try {
                    System.out.println("i am should print old ,but now is even:"+val+",so wait");
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }finally {
            lock.unlock();
        }
    }
    public void waitToEven(){
        lock.lock(); //显示的锁定
        try{
            while ((val % 2)!=0){
                try {
                    System.out.println("i am should print even ,but now old:"+val+",so wait");
                    condition.await();//执行等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }finally {
            lock.unlock(); //显示的释放
        }

    }
}
复制代码

同样可以得到上述的效果

显示锁的功能

显示锁在java中通过接口Lock提供如下功能

• lock: 线程无法获取锁会进入休眠状态，直到获取成功
• lockInterruptibly: 如果获取成功，立即返回，否则一直休眠到线程被中断或者是获取成功
• tryLock：不会造成线程休眠，方法执行会立即返回，获取到了锁，返回true,否则返回false
• tryLock(long time, TimeUnit unit) throws InterruptedException : 在等待时间内没有发生过中断，并且没有获取锁，就一直等待，当获取到了，或者是线程中断了，或者是超时时间到了这三者发生一个就返回，并记录是否有获取到锁
• unlock：释放锁
• newCondition：每次调用创建一个锁的等待条件，也就是说一个锁可以拥有多个条件

Condition的功能

接口Condition把Object的监视器方法wait和notify分离出来，使得一个对象可以有多个等待的条件来执行等待，配合Lock的newCondition来实现。

• await:使当前线程休眠，不可调度。这四种情况下会恢复 1：其它线程调用了signal，当前线程恰好被选中了恢复执行;2: 其它线程调用了signalAll;3:其它线程中断了当前线程 4：spurious wakeup (假醒)。无论什么情况，在await方法返回之前，当前线程必须重新获取锁
• awaitUninterruptibly:使当前线程休眠，不可调度。这三种情况下会恢复 1：其它线程调用了signal，当前线程恰好被选中了恢复执行;2: 其它线程调用了signalAll;3：spurious wakeup (假醒)。
• awaitNanos:使当前线程休眠，不可调度。这四种情况下会恢复 1：其它线程调用了signal，当前线程恰好被选中了恢复执行;2: 其它线程调用了signalAll;3:其它线程中断了当前线程 4：spurious wakeup (假醒)。5:超时了
• await(long time, TimeUnit unit) ：与awaitNanos类似，只是换了个时间单位
• awaitUntil(Date deadline):与awaitNanos相似，只是指定日期之后返回，而不是指定的一段时间
• signal:唤醒一个等待的线程
• signalAll:唤醒所有等待的线程

ReentrantLock

从源码中可以看到，ReentrantLock的所有实现全都依赖于内部类Sync和ConditionObject。

Sync本身是个抽象类，负责手动lock和unlock,ConditionObject则实现在父类AbstractOwnableSynchronizer中，负责await与signal Sync的继承结构如下

Sync的两个实现类，公平锁和非公平锁

公平的锁会把权限给等待时间最长的线程来执行,非公平则获取执行权限的线程与线程本身的等待时间无关

public ReentrantLock() {
   sync = new NonfairSync();
}
复制代码

当执行获取锁时，实际就是去执行 Sync 的lock操作：

public void lock() {
    sync.lock();
}
复制代码

对应在不同的锁机制中有不同的实现

1. 公平锁实现

   final void lock() {
       acquire(1);
   }
   复制代码

2. 非公平锁实现

   final void lock() {
       if (compareAndSetState(0, 1)) //先看当前锁是不是已经被占有了，如果没有，就直接将当前线程设置为占有的线程
           setExclusiveOwnerThread(Thread.currentThread());
       else        
           acquire(1); //锁已经被占有的情况下，尝试获取
   }
   复制代码

二者都调用父类AbstractQueuedSynchronizer的方法

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //一旦抢失败，就会进入队列，进入队列后则是依据FIFO的原则来执行唤醒
        selfInterrupt();
}
复制代码

当执行unlock时，对应方法在父类AbstractQueuedSynchronizer中

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
复制代码

公平锁和非公平锁则分别对获取锁的方式 tryAcquire 做了实现，而tryRelease的实现机制则都是一样的

公平锁实现tryAcquire

源码如下

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState(); //获取当前的同步状态
    if (c == 0) {
        //等于0 表示没有被其它线程获取过锁
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            //hasQueuedPredecessors 判断在当前线程的前面是不是还有其它的线程，如果有，也就是锁sync上有一个等待的线程，那么它不能获取锁，这意味着，只有等待时间最长的线程能够获取锁，这就是是公平性的体现
            //compareAndSetState 看当前在内存中存储的值是不是真的是0，如果是0就设置成accquires的取值。对于JAVA，这种需要直接操作内存的操作是通过unsafe来完成，具体的实现机制则依赖于操作系统。
            //存储获取当前锁的线程
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        //判断是不是当前线程获取的锁
        int nextc = c + acquires;
        if (nextc < 0)//一个线程能够获取同一个锁的次数是有限制的，就是int的最大值
            throw new Error("Maximum lock count exceeded");
        setState(nextc); //在当前的基础上再增加一次锁被持有的次数
        return true;
    }
    //锁被其它线程持有，获取失败
    return false;
}
复制代码

非公平锁实现tryAcquire

获取的关键实现为 nonfairTryAcquire ,源码如下

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        //锁没有被持有
        //可以看到这里会无视sync queue中是否有其它线程，只要执行到了当前线程，就会去获取锁
        if (compareAndSetState(0, acquires)) { 
            setExclusiveOwnerThread(current); //在判断一次是不是锁没有被占有，没有就去标记当前线程拥有这个锁了
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires; 
        if (nextc < 0) // overflow            
            throw new Error("Maximum lock count exceeded");
        setState(nextc);//如果当前线程已经占有过，增加占有的次数
        return true;
    }
    return false;
}
复制代码

释放锁的机制

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread()) //只能是线程拥有这释放
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        //当占有次数为0的时候，就认为所有的锁都释放完毕了
        free = true; 
        setExclusiveOwnerThread(null);
    }
    setState(c); //更新锁的状态
    return free;
}
复制代码

从源码的实现可以看到

ReetrantLock本身对锁的持有是可重入的,同时是线程独占的

public boolean tryLock() {
//本质上就是执行一次非公平的抢锁
   return sync.nonfairTryAcquire(1); 
}
复制代码

AbstractQueuedSynchronizer

无论是公平锁还是非公平锁，它们的实现都依赖于AbstractQueuedSynchronizer，它提供了一个基于先进先出等待队列 实现block locks和synchronizers的框架。特性如下

• 仅通过一个 int 类型来代表状态。对于ReentrantLock而言，他就是线程持有锁的次数，当次数为0时，代表锁没有被持有，正数代表被持有的次数，负数则是超出了锁的持有范围，有可能存在死循环
• 支持独占模式(默认的)和共享模式。在独占模式中，去获取一个已经被其它线程拥有的锁只会失败，共享模式中，则多个线程是可以成功的。

lock()原理

当ReentrantLock获取锁失败时，会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

private Node addWaiter(Node mode) { 
    Node node = new Node(Thread.currentThread(), mode); //创建一个节点，存储当前的线程，以及锁持有的模式，对于 ReentrantLock来说就是 独占 型
    // Try the fast path of enq; backup to full enq on failure    
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {//CAS操作，如果当前的尾部节点没有被其它线程更改，那么把新的节点设置成队列的尾部
            pred.next = node;
            return node;
        }
    }
    enq(node);//首次入队
    return node;
}
复制代码

获取失败进行入队操作，首先就是往队列中添加一个正在等待的节点Node

从Node本身的结构可以看到，AQS(AbstractQueuedSynchronizer)本身就维护了一个双向链表，用来存放等待中的线程。链表的每个节点，代表那个线程，是独占还是共享锁。

创建好节点之后，便执行入队操作，对于首次创建队列

private Node enq(final Node node) {
    for (;;) {
       //借助CAS机制实现无锁操作，所以需要一直执行直到CAS成功
        Node t = tail;
        if (t == null) { // 初始化发生在第一次创建队列，这样的好处是，当竞争不激烈的时候，实际上也就不会发生这些操作，性能也会好些
             if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
复制代码

可以看到，入队也就是从队尾插入新的等待线程，入队完毕，也就开始去进行不断的尝试，直到获取锁成功,可以看到，对于lock来说，其实已经是阻塞了

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor(); //优先执行当前节点的前一个节点
            if (p == head && tryAcquire(arg)) { //仅当当前节点的前一个节点是head,才去获取线程，这里可以看出其实先等待的线程是会优先处理，也就是FIFO原则
                setHead(node); 
                p.next = null; // help GC    ,释放掉当前线程在队列中的引用，也可以看做’出队'了            
                failed = false;
                //执行到这里说明获取锁成功
                return interrupted;
            }
            //执行到这里说明存在竞争，有多个线程都在等待一个锁
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt()) //里面会对当前线程执行中断，当被唤醒时，继续循环
                    //如果线程被中断，设置中断标记，区别于 doAcquireInterruptibly，doAcquireInterruptibly是直接抛出异常，这也就是 lockInterruptibly能够抛出中断的原因
                    interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
复制代码

从这里可以看到，无论锁是公平锁还是非公平锁，只要被放入了等待队列，此时的执行依然是谁先等待就先执行谁 ,非公平锁体现在新来的线程会无视已经等了的线程，可以优先去抢锁,所以公平体现在第一次参与抢锁的线程会去等待已经在等待队列中的线程,非公平并不是说从已经在等待的线程队列里面随便选一个

shouldParkAfterFailedAcquire的源码如下

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus; //查看前一个节点的等待状态
    if (ws == Node.SIGNAL)
            //已经尝试过获取锁，可以执行park了
            return true;
    if (ws > 0) {   
        do {
            //去掉队列中所有已经取消的线程
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {             
        //此时当前线程的前一个节点的等待状态必定是0或者PROGATE，这表明当前线程在park之前可以再尝试一次去获取锁,也就是说前一个节点可能刚获取到SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

复制代码

waitStatus：等待的状态，共有5种

• SIGNAL:，表明它的前一个节点需要执行 unparking；
• CANCELLED：当前节点保存的线程由于超时或者中断被取消了；
• CONDITION：接档正处于条件队列中，执行了await；
• PROPAGATE：一个共享的锁需要传递释放信号到其它节点
• 0:非上述4中状态，有可能是刚获取signal,此时它的值是0，也有可能是新建的head节点

parkAndCheckInterrupt主要是park当前线程

private final boolean parkAndCheckInterrupt() {
    //当获取不到许可时，阻塞线程，解除阻塞状态的情况如下：
    //1 某个线程对这个线程调用了unpark方法
    //2 某个线程中断了这个线程
    //3 这个方法毫无理由的返回了 [park比较奇特的地方],基于这样，调用的时候必须去判断park的条件，以及当它返回的时候，去设置中断的状态
    LockSupport.park(this); 
    //返回线程的中断状态
    return Thread.interrupted();
}
复制代码

至此lock()执行结束

unlock()原理

当执行unlock时，ReentrentLock执行对应的Release

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        //执行这里表示所已经被释放,可以让它的下一个节点来抢锁
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h); //h.waitStatus == 0 表示还没有执行park,自然不需要unpark
        return true;
    }
    return false;
}
复制代码

如果release成功，即当前线程持有的所有锁都已经释放，那么就可以执行 unparkSuccessor ,从源码可以看到，unpark是从头部开始进行的，结合lock的原理，可知AQS本身就是一个先进先出的队列 unparkSuccessor源码如下

private void unparkSuccessor(Node node) {
     int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

     Node s = node.next; //有可能线程被取消了，所以从尾部往前遍历，到最近一个没有被取消的线程
    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; //确保线程没有取消
    }
    if (s != null)
        LockSupport.unpark(s.thread); //恢复线程
}
复制代码

至此unlock()完毕

await的原理

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();//当前线程已经中断了，抛出中断异常
    //添加一个新的waiter到condition queue中,这个新的Node的waitStatus会被标记为CONDITION
    Node node = addConditionWaiter(); 
    //释放当前线程拥有的锁,即从sync queue中去掉当前线程
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
    //如果当前线程不在持有锁的队列里头,对他进行休眠，当其它线程执行 unlock的时候，释放锁，就会执行unpark操作，此时它会被唤醒，唤醒后，如果它在syn队列里头，开始继续往下执行。(这个插入操作则是由signal完成)
        LockSupport.park(this); 
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;//等待的过程中线程中断了，退出
    }
//重新竞争锁,相当于执行了lock操作
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    //再次去获取锁，如果当前的线程在park的时候是被中断了，并且ConditionObject并不是由于中断返回，这里再次标记为中断
        interruptMode = REINTERRUPT;  
    if (node.nextWaiter != null) 
    //清除非Condition模式的线程,而在signal中有先关操作将conditon的线程设置成非condition
         unlinkCancelledWaiters();
    if (interruptMode != 0)
    //上报等待的过程中发生了中断,如果是要抛出中断，就抛出，否则再次执行中断
        reportInterruptAfterWait(interruptMode); 
}
复制代码

isOnSyncQueue源码如下

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
    //node本身是调用了 await 方法,或者没有在获取锁的队列里头,[如果在里头必定有一个前置的节点]
        return false; 
    if (node.next != null) 
    //当前节点存在下一个节点，那么它肯定是执行过 enq ,即获取过锁
        return true; 
    // CAS失败的时候，有可能 node.rev是没有的，因此需要从头到尾遍历一次
   return findNodeFromTail(node); 
}
复制代码

checkInterruptWhileWaiting源码如下

private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    //线程中断重新获取锁,并且设置waitStatus为0，以便后续线程从condition queue清除
        enq(node); 
        return true;
    }
    while (!isOnSyncQueue(node))
    //如果CAS失败，只要当前节点没有在Sync queue中，那么一直自旋，每次都会交出执行权限
        Thread.yield(); 
    return false;
}
复制代码

可以看到，await其实就是释放线程原有的锁，并把它放入conditon队列中，然后执行阻塞。等唤醒的时候，重新获取锁，并清掉condition queue中的线程。 至此await执行结束

singnal的原理

public final void signal() {
    if (!isHeldExclusively()) //只有当前线程持有了锁，才能释放
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);//优先释放队列头的,也就是等待时间最长的condition node
}
private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}
//将节点从condition queue转移到sync queue
final boolean transferForSignal(Node node) {
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false; //设置为非等待失败，则不继续转移
//CAS设置等待状态为0成功
Node p = enq(node); //新节点放入sync queue，并返回原来的尾部节点，也就是新节点的前一个节点
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) //参考shouldParkAfterFailedAcquire
        LockSupport.unpark(node.thread);//如果当前节点的前一个节点线程已经取消,或者将当前节点的前一个节点线程的waitStatus设置成SIGNAL失败,则直接唤醒当前线程
    return true;
}
复制代码

可以看到signal最关键的信息就是去掉等待队列中的CONDITION状态，并将线程加入sync队列，至此signal结束