Java并发之AQS详解(二)源码解析

1 介绍

在Java并发之AQS详解(一)中，已经对AQS中主要的类，重点方法、流程进行了分析，本文针对一些重点的方法逻辑进行源码层面的解读分析。不对的地方欢迎大家指正交流。

• 独占方式的获取与释放资源
• 分享方式的获取去释放资源
• 入队与出队源码进行分析

2 图示队列

3 方法分析

3.1 独占方式

3.1 acquire(int arg)获取共享资源，忽略中断

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
           

代码运行流程

1. tryAcquire()尝试直接去获取资源，如果成功则直接返回；
2. addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
3. acquireQueued()使线程阻塞在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
4. 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。

3.1.1 tryAcquire(arg)

尝试去获取独占资源。如果获取成功，则直接返回true，否则直接返回false。

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
}
           

我们通过这一段源码就可以知道前面我们说过的，AQS只是一个框架，具体资源的获取交由自定义同步器去实现，能不能重入，是否可以加塞(公平、非公平)就看具体的同步器怎么实现了。我们可以看一下ReentrantLock类中tryAcquire(arg) 方法的具体实现

protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            //查看资源是否有其他线程已占用 当c==0时没有被占用
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            //是否为当前线程再一次获取资源
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
           

3.1.2 addWaiter(Node mode)

将当前线程加入到等待队列的队尾，并返回当前线程所在的结点

private Node addWaiter(Node mode) {
 		//以给定模式构造结点。mode有两种：EXCLUSIVE（独占）和SHARED（共享）
        Node node = new Node(Thread.currentThread(), mode);
        // 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)) {
                pred.next = node;
                return node;
            }
        }
        //如果上一步失败，则通过enq入队
        enq(node);
        return node;
    }
           

3.1.3 enq(node)

enq(final Node node） 方法将该节点插入到AQS 的阻塞队列

private Node enq(final Node node) {
		// CAS自旋直到成功加入队尾
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize 初始化: 创建一个空的标志结点作为head结点，并将tail也指向它。
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
            //正常加入队列中
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
           

3.1.4 acquireQueued(Node, int)

通过tryAcquire()和addWaiter()，该线程获取资源失败，已经被放入等待队列尾部了。当前线程现在会进入等待状态休息，直到其他线程彻底释放资源后唤醒自己，自己再拿到资源，完成自己的逻辑。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;//标记是否成功拿到资源
    try {
        boolean interrupted = false;//标记等待过程中是否被中断过

        //自旋
        for (;;) {
            final Node p = node.predecessor();//拿到前驱
            //如果前驱是head，该结点已有资格去尝试获取资源
            if (p == head && tryAcquire(arg)) {
                setHead(node);//拿到资源后，将head指向该结点。所以head所指的标杆结点，就是当前获取到资源的那个结点或null。
                p.next = null; //此处再将head.next置为null，就是为了方便GC回收以前的head结点。 help GC
                failed = false; // 成功获取资源
                return interrupted;//返回等待过程中是否被中断过
            }

            //如果还不是上述的情况，就通过park()进入waiting状态，直到被unpark()。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;//如果等待过程中被中断过，就将interrupted标记为true
        }
    } finally {
        if (failed) // 如果等待过程中出现异常，那么取消结点在队列中的等待。
            cancelAcquire(node);
    }
}
           

3.2 release(int)

独占模式下线程释放共享资源。释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;//找到头结点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);//唤醒等待队列里的下一个线程
        return true;
    }
    return false;
}
           

当一个线程调用release(int arg）方法时会尝试使用tryRelease 操作释放资源，这里是设置状态变量state 的值，然后调用LockSupport.unpark(thread）方法激活AQS 队列里面被阻塞的一个线程(thread)。被激活的线程则使用tryAcquire 尝试，看当前状态变量state的值是否能满足自己的需要，满足则该线程被激活，然后继续向下运行，否则还是会被放入AQS 队列并被挂起。

文章开头已经说过AQS 类并没有提供可用的tryAcquire 和tryRelease 方法， tryAcquire 和tryRelease 需要由具体的子类来实现。上面已经分析了lock实现的tryAcquire方法的实现，在这不再源码分析tryRelease，感兴趣的小伙伴可以自己研究一下。

3.2.1 unparkSuccessor(Node)

唤醒等待队列中下一个线程

private void unparkSuccessor(Node node) {
    //这里，node一般为当前线程所在的结点。
    int ws = node.waitStatus;
    if (ws < 0)//置零当前线程所在的结点状态，允许失败。
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;//找到下一个需要唤醒的结点s
    if (s == null || s.waitStatus > 0) {//如果为空或已取消
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev) // 从后向前找。
            if (t.waitStatus <= 0)//从这里可以看出，<=0的结点，都是还有效的结点。
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);//唤醒
}
           

4 总结

此文只分析了独占模式下的各个方法，接下来一篇文章会从源码层分析一下共享模式下的重点方法。