Java源码分析----ReentrantLock实现和Abst

Java并发包下很多组件的基础都是AQS，例如ReentrantLock，CountDownLatch等等，只要明白了AQS的原理，其他的基本可以理解，那么就从ReentrantLock为例，进行分析

AQS中有个state的状态，以Lock为例，如果哪个线程能成功将state设置为1，那么证明该线程获取到锁

ReentrantLock分为公平锁和非公锁，当时用下面的方式创建的lock是非公平锁

ReentrantLock lock = new ReentrantLock();

那么，先看下ReentrantLock中有什么东西

public class ReentrantLock implements Lock, java.io.Serializable {
    private final Sync sync;
    abstract static class Sync extends AbstractQueuedSynchronizer {

        final boolean nonfairTryAcquire(int acquires) {
            ....
        }

        protected final boolean tryRelease(int releases) {
            ....
        }
    }

    static final class NonfairSync extends Sync {
        final void lock() {
            ....
        }

        protected final boolean tryAcquire(int acquires) {
            ....
        }
    }

    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;
        final void lock() {
            ....
        }
        protected final boolean tryAcquire(int acquires) {
            ....
        }
    }
    public ReentrantLock() {
        sync = new NonfairSync();
    }
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
}

其中有一个Sync类型的变量，分别有公平锁FairSync和非公平锁NonfairSync的实现，默认是非公平锁，而Sync的又继承于AQS，核心操作都是在这里面的，下面先以NonfairSync的lock方法开始分析

    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

lock的时候，会先调用compareAndSetState方法设置状态(state)值，如果成功，那么将持有锁的线程设置为当前线程

compareAndSetState方法如下：

    protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

非常粗暴，直接使用CAS操作，当置换成功就代表持有锁，即使是后来的线程，也可能会比先来的线程先持有锁，难怪叫非公平锁
setExclusiveOwnerThread就是简单的将exclusiveOwnerThread变量设置为currentThread

那么获取锁失败会进行什么操作了，如果多个线程都失败了呢，猜一下就是应该有个结构去保存这些失败的线程，当取的锁的线程释放后，这些线程再去竞争，那么acquire就是进行这样的操作

    //有个int类型的参数，这个参数在Lock里只是1，而信号量则是设置的信号值，也就是state的值
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

改方法会调用tryAcquire尝试获取锁，该方法由子类实现，非公平锁会调用Sync类的nonfairTryAcquire方法

    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            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;
    }

1. 如果state等于0，表示此刻没有线程获取到锁
   2.1 是CAS替换state值，如果成功，设置持有锁的线程，如果失败，代表其他线程在这个时候获取到了
2. 如果获取到锁的线程为当前线程，那么将state+1(为什么要加1，是因为Lock是可重入锁，那么就可能lock多次，state就+多次，解锁的时候，需要unlock多次，直到0，才算释放锁)
3. 否则，已经有线程持有锁，返回false表示失败

当失败的时候，就会执行下面的代码

acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

那么，我们猜，这里就是保存操作失败的线程数据的地方了，内部是使用了一个链表的结构去保存竞争失败的线程，每个节点的结构就是Node

    static final class Node {
        static final Node SHARED = new Node();
        static final Node EXCLUSIVE = null;
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;

        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;

        volatile Thread thread;

        Node nextWaiter;
}

学过数据结构的对这个结构就很熟悉了，那么节点如何进入链表里的，要看addWaiter方法

    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

首先会构建一个节点，Node，获取尾指针，由于第一次进行，tail为空，那么会执行enq方法

    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

tail为空，那么将一个空节点设置为头节点，且head和tail都指向这个节点，此时结构如下

当链表中没有元素的时候的结构.png

之后，进入第二次循环，t 这时也执行了头节点，且新增的节点的prev执行头节点，执行后就是下面的样子

image.png

再看下addWaiter节点就是往尾部增加节点，节点进入链表后，会调用acquireQueued方法

    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)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

假如此刻有一个线程获取到锁，2个线程进入等待，链表的结构如下

image.png

此时，AB都竞争失败入了链表中，假设B先执行该方法，发现它的前驱(node.predecessor())不是头节点，那么执行shouldParkAfterFailedAcquire方法

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        // ws>0：状态为CANCELLED
        // 找到最前面状态不是CANCELLED的节点，把这之间的节点CANCELLED断开        
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            //  如果前面的节点为正常状态
            // 将前面节点的状态设置为SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

该方法返回是否需要挂起线程，看下其中的判断

1. 当前节点状态等于SIGNAL，此时需要挂起线程
2. 如果状态>0(表示CANCELLED)，那么往前找到一个不是CANCELLED状态的节点，且把CANCELLED状态的节点断开
3. 如果上面两个条件不满足，那么将状态设置为SIGNAL然后返回false表示不需要挂起

此时B会重新进入循环判断，前驱仍然不是头节点，但是前驱节点的状态为SIGNAL，那么parkAndCheckInterrupt方法将挂起该线程

再看下A，它的前驱是头节点，那么如果tryAcquire尝试获取锁，如果获取锁的线程还没释放锁，那么和B一样，执行上述流程，如果已经释放了锁，那么tryAcquire返回true，先调用setHead方法

    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

将A设置为头节点，Node中的信息置空，看到这里，也就是说将A变成了上面new Node一样的状态，那么我们明白了，头节点这个空节点也就是代表一个持有锁的节点

到这里，A已经获取锁了，可以执行他的代码了，B仍然在挂起状态，直到A释放锁

那么看下释放锁的流程，unLock会调用到AQS的release方法

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

tryRelease在Sync中实现，和tryacquire的过程相反这里就不说了，当tryRelease成功后，调用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);
    }

1. 首先将节点状态设置为0
2. 获取该节点后的第一个节点，如果节点不为空且状态不是取消状态，那么唤起该线程
3. 如果后续节点为空，或者状态为取消状态，那么从tail往前找到第一个消状态小于等于0的节点

当唤醒成功，即解锁成功，那么刚刚B会从parkAndCheckInterrupt中继续运行，然后重新进入for循环，此时，他的前驱节点是head，且已经释放锁，那么tryacquire也可以成功，之后进行和A一样的操作

看到这里ReentrantLock和AQS的实现就分析完了，上面分析AB的情况只是其中一种简单的情况，在并发的时候，可能顺序并不这样或者节点比较多，分析的情况比较多，不过原理差不多，明白这一种情况后去分析其他的就比较好理解了

注：AQS还有共享锁等的其他实现，这个以后有空会针对某个组件进行分析