【JDK】JDK源码分析-AbstractQueuedSynchronizer(1)

概述

前文「JDK源码分析-Lock&Condition」简要分析了 Lock 接口，它在 JDK 中的实现类主要是 ReentrantLock (可译为“重入锁”)。ReentrantLock 的实现主要依赖于其内部的一个嵌套类 Sync，而 Sync 又继承自 AbstractQueuedSynchronizer (简称 AQS)。而且，不仅 ReentrantLock，其他一些并发工具类如 CountdownLatch、CyclicBarrier 等，其实现也都是基于 AQS 类。AQS 可以理解为并发包中许多类实现的基石。因此，在分析并发包中常用类的实现原理前，有必要先理解一下 AQS，之后再分析的时候就会简单不少。

AQS 内部有一个核心变量 state；此外，以 Node 类为节点维护了两种队列：主队列（main queue）和条件队列（condition queue），简单起见，分别可以将二者理解为双链表和单链表。

AQS 就像是提供了一套基础设施的设备，其它常用类如 ReentrantLock、CountdownLatch 等的内部嵌套类 Sync，都是在 AQS 提供的基础设施之上制定了自己的“游戏规则”，进而生产出了不同的产品。而它们的游戏规则都是围绕 state 变量和这两种队列进行操作的。

PS: 由于 AQS 内容较多，因此打算分多篇文章进行分析，本文先对其整体进行概述。

代码分析

AQS 类签名：

public abstract class AbstractQueuedSynchronizer

    extends AbstractOwnableSynchronizer

    implements java.io.Serializable {}

可以看到它是一个抽象类，不能直接被实例化。它的父类 AbstractOwnableSynchronizer 的主要代码如下：

public abstract class AbstractOwnableSynchronizer

    implements java.io.Serializable {

    /**

     * The current owner of exclusive mode synchronization.

     */

    private transient Thread exclusiveOwnerThread;

    // 其他代码

}

其内部主要维护了一个变量 exclusiveOwnerThread，作用是标记独占模式下的 Owner 线程，后面涉及到的时候再进行分析。

嵌套类

AQS 内部有两个嵌套类，分别为 Node 和 ConditionObject。

Node 类代码如下：

static final class Node {

    // 共享模式

    static final Node SHARED = new Node();

    // 独占模式

    static final Node EXCLUSIVE = null;

    // waitStatus的几种状态

    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;

    final boolean isShared() {

        return nextWaiter == SHARED;

    }

    final Node predecessor() throws NullPointerException {

        Node p = prev;

        if (p == null)

            throw new NullPointerException();

        else

            return p;

    }

    Node() {    // Used to establish initial head or SHARED marker

    }

    Node(Thread thread, Node mode) {     // Used by addWaiter

        this.nextWaiter = mode;

        this.thread = thread;

    }

    Node(Thread thread, int waitStatus) { // Used by Condition

        this.waitStatus = waitStatus;

        this.thread = thread;

    }

}

添加到主队列用的是第二个构造器，Node 类可以理解为对线程 Thread 的封装。因此，在主队列中排队的一个个节点可以理解为一个个有模式（mode）、有状态（waitStatus）的线程。

嵌套类 ConditionObject：

public class ConditionObject implements Condition, java.io.Serializable {

    /** First node of condition queue. */

    private transient Node firstWaiter;

    /** Last node of condition queue. */

    private transient Node lastWaiter;

    // ...

}

ConditionObject 实现了 Condition 接口，它主要操作的是条件队列，这里只贴了其类签名和头尾节点，后面用到的时候再具体分析。

主要变量

AQS 代码虽长，但它的成员变量却不多，如下：

// 主队列头节点

private transient volatile Node head;

// 主队列尾结点

private transient volatile Node tail;

// 状态，AQS 维护的一个核心变量

private volatile int state;

其中，head 和 tail 为主队列的头尾节点，state 为 AQS 维护的核心变量，ReentrantLock 等类中的 Sync 类实现，都是通过操作 state 来实现各自功能的。

CAS 操作

AQS 内部通过 Unsafe 类实现了一系列 CAS (Compare And Swap) 操作（有关 CAS 的概念这里不再详解，可自行搜索了解）：

// 获取 Unsafe 实例

private static final Unsafe unsafe = Unsafe.getUnsafe();

// state、head、tail 等变量的内存偏移地址

private static final long stateOffset;

private static final long headOffset;

private static final long tailOffset;

private static final long waitStatusOffset;

private static final long nextOffset;

static {

    try {

        stateOffset = unsafe.objectFieldOffset

            (AbstractQueuedSynchronizer.class.getDeclaredField("state"));

        headOffset = unsafe.objectFieldOffset

            (AbstractQueuedSynchronizer.class.getDeclaredField("head"));

        tailOffset = unsafe.objectFieldOffset

            (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));

        waitStatusOffset = unsafe.objectFieldOffset

            (Node.class.getDeclaredField("waitStatus"));

        nextOffset = unsafe.objectFieldOffset

            (Node.class.getDeclaredField("next"));

    } catch (Exception ex) { throw new Error(ex); }

}

// 一些 CAS 操作

private final boolean compareAndSetHead(Node update) {

    return unsafe.compareAndSwapObject(this, headOffset, null, update);

}

private final boolean compareAndSetTail(Node expect, Node update) {

    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);

}

private static final boolean compareAndSetWaitStatus(Node node,

                                                     int expect,

                                                     int update) {

    return unsafe.compareAndSwapInt(node, waitStatusOffset,

                                    expect, update);

}

private static final boolean compareAndSetNext(Node node,

                                               Node expect,

                                               Node update) {

    return unsafe.compareAndSwapObject(node, nextOffset, expect, update);

}

AQS 内部的许多操作是通过 CAS 来实现线程安全的。

小结

1. AQS 是一个抽象类，无法直接进行实例化；

2. AQS 内部维护了一个核心变量 state，以及两种队列：主队列（main queue）和条件队列（condition queue）；

3. AQS 提供了一套基础设施，ReentrantLock 等类通常用一个内部嵌套类 Sync 继承 AQS，并在 Sync 类中制定自己的“游戏规则”。

本文仅对 AQS 做了概述，后面再详细分析实现原理。此外，还有一个类 AbstractQueuedLongSynchronizer，它与 AQS 基本完全一样，区别在于前者的 state 变量为 long 类型，而 AQS 为 int 类型，不再单独进行分析。

PS: 有几篇文章写得也不错，链接如下：

https://www.cnblogs.com/liuyun1995/p/8400663.html

Stay hungry, stay foolish.

PS: 本文首发于微信公众号【WriteOnRead】。

巴特西

【JDK】JDK源码分析-AbstractQueuedSynchronizer(1)

最新文章

热门文章