技术 · 阅读记录

Hello, AQS(AbstractQueuedSynchronizer)

fxleyu2018年9月14日

未有特殊说明，本文源码来源于 jdk1.6.0_45。

AbstractQueuedSynchronizer 为实现依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关同步器（信号量、事件，等等）提供一个框架。此类的设计目标是成为依靠单个原子 int 值来表示状态的大多数同步器的一个有用基础。

子类必须定义更改此状态的受保护方法，并定义哪种状态对于此对象意味着被获取或被释放。假定这些条件之后，此类中的其他方法就可以实现所有排队和阻塞机制。子类可以维护其他状态字段，但只是为了获得同步而只追踪使用 getState()、setState(int) 和 compareAndSetState(int, int) 方法来操作以原子方式更新的 int 值。

应该将子类定义为非公共内部帮助器类，可用它们来实现其封闭类的同步属性。类 AbstractQueuedSynchronizer 没有实现任何同步接口。而是定义了诸如 acquireInterruptibly(int) 之类的一些方法，在适当的时候可以通过具体的锁和相关同步器来调用它们，以实现其公共方法。

此类支持默认的独占模式和共享模式之一，或者二者都支持。处于独占模式下时，其他线程试图获取该锁将无法取得成功。在共享模式下，多个线程获取某个锁可能（但不是一定）会获得成功。此类并不“了解”这些不同，除了机械地意识到当在共享模式下成功获取某一锁时，下一个等待线程（如果存在）也必须确定自己是否可以成功获取该锁。处于不同模式下的等待线程可以共享相同的 FIFO 队列。通常，实现子类只支持其中一种模式，但两种模式都可以在（例如）ReadWriteLock 中发挥作用。只支持独占模式或者只支持共享模式的子类不必定义支持未使用模式的方法。

此类为内部队列提供了检查、检测和监视方法，还为 condition 对象提供了类似方法。可以根据需要使用用于其同步机制的 AbstractQueuedSynchronizer 将这些方法导出到类中。

此类的序列化只存储维护状态的基础原子整数，因此已序列化的对象拥有空的线程队列。需要可序列化的典型子类将定义一个 readObject 方法，该方法在反序列化时将此对象恢复到某个已知初始状态。

使用

为了将此类用作同步器的基础，需要适当地重新定义以下方法，这是通过使用 getState()、 setState(int) 和/或 compareAndSetState(int, int) 方法来检查和/或修改同步状态来实现的：

tryAcquire(int)
tryRelease(int)
tryAcquireShared(int)
tryReleaseShared(int)
isHeldExclusively()

默认情况下，每个方法都抛出 UnsupportedOperationException。这些方法的实现在内部必须是线程安全的，通常应该很短并且不被阻塞。定义这些方法是使用此类的唯一受支持的方式。其他所有方法都被声明为 final，因为它们无法是各不相同的。

您也可以查找从 AbstractOwnableSynchronizer 继承的方法，用于跟踪拥有独占同步器的线程。鼓励使用这些方法，这允许监控和诊断工具来帮助用户确定哪个线程保持锁。

即使此类基于内部的某个 FIFO 队列，它也无法强行实施 FIFO 获取策略。独占同步的核心采用以下形式：

//Acquire:
    while (!tryAcquire(arg)) {
       enqueue thread if it is not already queued;
       possibly block current thread;
    }

//Release:
    if (tryRelease(arg))
       unblock the first queued thread;

（共享模式与此类似，但可能涉及级联信号。）因为要在加入队列之前检查线程的获取状况，所以新获取的线程可能闯入其他被阻塞的和已加入队列的线程之前。不过如果需要，可以内部调用一个或多个检查方法，通过定义 tryAcquire 和/或 tryAcquireShared 来禁用闯入。特别是 getFirstQueuedThread() 没有返回当前线程的时候，严格的 FIFO 锁定可以定义 tryAcquire 立即返回 false。只有 hasQueuedThreads() 返回 true 并且 getFirstQueuedThread 不是当前线程时，更好的非严格公平的版本才可能会立即返回 false；如果 getFirstQueuedThread 不为 null 并且不是当前线程，则产生的结果相同。出现进一步的变体也是有可能的。

对于默认闯入（也称为 greedy、renouncement 和 convoy-avoidance）策略，吞吐量和可伸缩性通常是最高的。尽管无法保证这是公平的或是无偏向的，但允许更早加入队列的线程先于更迟加入队列的线程再次争用资源，并且相对于传入的线程，每个参与再争用的线程都有平等的成功机会。此外，尽管从一般意义上说，获取并非“自旋”，它们可以在阻塞之前对用其他计算所使用的 tryAcquire 执行多次调用。在只保持独占同步时，这为自旋提供了最大的好处，但不是这种情况时，也不会带来最大的负担。如果需要这样做，那么可以使用“快速路径”检查来先行调用 acquire 方法，以这种方式扩充这一点，如果可能不需要争用同步器，则只能通过预先检查 hasContended() 和/或 hasQueuedThreads() 来确认这一点。

通过特殊化其同步器的使用范围，此类为部分同步化提供了一个有效且可伸缩的基础，同步器可以依赖于 int 型的 state、acquire 和 release 参数，以及一个内部的 FIFO 等待队列。这些还不够的时候，可以使用 atomic 类、自己的定制 Queue 类和 LockSupport 阻塞支持，从更低级别构建同步器。

使用示例

以下是一个非再进入的互斥锁类，它使用值 0 表示未锁定状态，使用 1 表示锁定状态。当非重入锁定不严格地需要当前拥有者线程的记录时，此类使得使用监视器更加方便。它还支持一些条件并公开了一个检测方法：

 class Mutex implements Lock, java.io.Serializable {

    // Our internal helper class
    private static class Sync extends AbstractQueuedSynchronizer {
      // Report whether in locked state
      protected boolean isHeldExclusively() {
        return getState() == 1;
      }

      // Acquire the lock if state is zero
      public boolean tryAcquire(int acquires) {
        assert acquires == 1; // Otherwise unused
       if (compareAndSetState(0, 1)) {
         setExclusiveOwnerThread(Thread.currentThread());
         return true;
       }
       return false;
      }

      // Release the lock by setting state to zero
      protected boolean tryRelease(int releases) {
        assert releases == 1; // Otherwise unused
        if (getState() == 0) throw new IllegalMonitorStateException();
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
      }

      // Provide a Condition
      Condition newCondition() { return new ConditionObject(); }

      // Deserialize properly
      private void readObject(ObjectInputStream s)
        throws IOException, ClassNotFoundException {
        s.defaultReadObject();
        setState(0); // reset to unlocked state
      }
    }

    // The sync object does all the hard work. We just forward to it.
    private final Sync sync = new Sync();

    public void lock()                { sync.acquire(1); }
    public boolean tryLock()          { return sync.tryAcquire(1); }
    public void unlock()              { sync.release(1); }
    public Condition newCondition()   { return sync.newCondition(); }
    public boolean isLocked()         { return sync.isHeldExclusively(); }
    public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
    public void lockInterruptibly() throws InterruptedException {
      sync.acquireInterruptibly(1);
    }
    public boolean tryLock(long timeout, TimeUnit unit)
        throws InterruptedException {
      return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
 }

以下是一个锁存器类，它类似于 CountDownLatch，除了只需要触发单个 signal 之外。因为锁存器是非独占的，所以它使用 shared 的获取和释放方法。

 class BooleanLatch {

    private static class Sync extends AbstractQueuedSynchronizer {
      boolean isSignalled() { return getState() != 0; }

      protected int tryAcquireShared(int ignore) {
        return isSignalled()? 1 : -1;
      }

      protected boolean tryReleaseShared(int ignore) {
        setState(1);
        return true;
      }
    }

    private final Sync sync = new Sync();
    public boolean isSignalled() { return sync.isSignalled(); }
    public void signal()         { sync.releaseShared(1); }
    public void await() throws InterruptedException {
      sync.acquireSharedInterruptibly(1);
    }
 }

独占模式源码分析

0 `acquire(int)`

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 把当前线程设置为中断状态
        // Thread.currentThread().interrupt();
        selfInterrupt();
}

其中，tryAcquire(int) 方法是未支持方法，需要 AQS 的子类来实现。这里将详细刨析 AQS 本身实现的方法。

1 `addWaiter(Node.EXCLUSIVE)`

创建一个排它节点，并加入到队列的尾部。

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;
        }
    }

    // 失败，或者 tail 为 null
    enq(node);
    return node;
}

1.1 `enq(Node)`

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

2 `acquireQueued(Node, int)`

关起当前线程

final boolean acquireQueued(final Node node, int arg) {
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } catch (RuntimeException ex) {
        cancelAcquire(node);
        throw ex;
    }
}

3 `shouldParkAfterFailedAcquire(Node, Node)`

把其前置节点设置为 SIGNAL，暗示其后置节点需要被唤醒（unpark）

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 前置节点等待状态为 SIGNAL
        return true;
    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;
}

4 `parkAndCheckInterrupt()`

挂起当前线程，等待其他线程唤醒。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

5 `cancelAcquire(Node)`

private void cancelAcquire(Node node) {
    if (node == null)
        return;

    node.thread = null;

    // 把终止的节点移除队列
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // predNext is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary.
    Node predNext = pred.next;

    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
   node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

2 `release(int)`

资源释放。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        // h == null 代表没有需要释放的节点
        // h != null && h.waitStatus == 0 可能有些节点为初始化完成
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

其中，tryRelease(int) 方法是未支持方法，需要 AQS 的子类来实现。这里将详细刨析 AQS 本身实现的方法。

1 `unparkSuccessor(Node)`

唤醒该节点的后置节点中标识的线程。

private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling. It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    // 唤醒 node 可以被唤醒的线程
    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);
}

共享模式源码分析

0 `acquireShared(int)`

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

其中，tryAcquireShared(int) 方法是未支持方法，需要 AQS 的子类来实现。这里将详细刨析 AQS 本身实现的方法。

1 `doAcquireShared(int)`

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                interrupted = true;
        }
    } catch (RuntimeException ex) {
        cancelAcquire(node);
        throw ex;
    }
}

2 `setHeadAndPropagate(Node, int)`

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    /*
     * Try to signal next queued node if:
     * Propagation was indicated by caller,
     * or was recorded (as h.waitStatus) by a previous operation
     * (note: this uses sign-check of waitStatus because
     * PROPAGATE status may transition to SIGNAL.)
     * and
     * The next node is waiting in shared mode,
     * or we don't know, because it appears null
     *
     * The conservatism in both of these checks may cause
     * unnecessary wake-ups, but only when there are multiple
     * racing acquires/releases, so most need signals now or soon
     * anyway.
     */
    if (propagate > 0 || h == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

浏览全部文章

使用

使用示例

独占模式源码分析

0 acquire(int)

1 addWaiter(Node.EXCLUSIVE)

1.1 enq(Node)

2 acquireQueued(Node, int)

3 shouldParkAfterFailedAcquire(Node, Node)

4 parkAndCheckInterrupt()

5 cancelAcquire(Node)

2 release(int)

1 unparkSuccessor(Node)