xuMengqi's Blog

  1. 1. 背景
  2. 2. AbstractQueuedSynchronizer设计思想
    1. 2.1. 互斥量的维护
    2. 2.2. CLH队列结构
    3. 2.3. CLH控制流程
  3. 3. AbstractQueuedSynchronizer源码解析
    1. 3.1. acquire加锁过程
      1. 3.1.1. 尝试获取互斥量（tryAcquire）
      2. 3.1.2. 线程加入等待队列（addWaiter）
      3. 3.1.3. 重新尝试获取互斥量（acquireQueued）
      4. 3.1.4. 挂起线程（parkAndCheckInterrupt）
    2. 3.2. release释放锁过程
      1. 3.2.1. 尝试释放互斥量（tryRelease）
      2. 3.2.2. 唤醒后继节点（unparkSuccessor）
    3. 3.3. cancelAcquire取消加锁过程
  4. 4. ReentrantLock源码解析
    1. 4.1. 公平锁FairSync的加锁
    2. 4.2. 非公平锁NofairSync的加锁
    3. 4.3. 释放锁
> 并发编程 源码 多线程

AbstractQueuedSynchronizer和ReentrantLock是如何实现的？

作者：徐梦旗，发布于：2024-05-18 16:01，字数：3.1k，预计阅读：14分钟

本文基于 JDK1.8 编写。

1. 背景

相比于synchronized内置锁隐式地加锁和释放锁的方式，ReentrantLock则采用显式地加锁和释放锁的方式，如下代码：

private final ReentrantLock lock = new ReentrantLock();

public void m() {
    lock.lock();
    try {
        // ... method body     
    } finally {
        lock.unlock();
    }
}

相比于synchronized内置锁，ReentrantLock支持更多的操作，拥有更丰富的应用场景：

  • 支持响应中断的加锁：Lock#lockInterruptibly。
  • 支持非阻塞式的加锁：Lock#tryLock()。
  • 支持设置超时时间的加锁：Lock#tryLock(long, TimeUnit)。
  • 支持公平锁和非公平锁：ReentrantLock#ReentrantLock(boolean)。

从底层实现上看，ReentrantLock使用AbstractQueuedSynchronizer来完成锁的添加和释放。下面让我们一起来看AQS是如何实现的？

synchronized内置锁自引入锁升级后，加锁的性能已经和ReentrantLock不相上下，如果选用ReentrantLock，则更看重其更丰富的应用场景。

2. AbstractQueuedSynchronizer设计思想

由于AbstractQueuedSynchronizer设计较为复杂，笔者水平有限，这里仅讨论互斥模式添加锁和释放锁的设计。

2.1. 互斥量的维护

AQS将互斥量state的维护逻辑交由子类实现，也就是以下两个方法：

  • AbstractQueuedSynchronizer#tryAcquire：尝试获取互斥量。
  • AbstractQueuedSynchronizer#tryRelease：尝试释放互斥量。

2.2. CLH队列结构

AQS中维护了一个CLH等待队列，队列中各节点Node由以下几部分组成：

  • thread：关联的线程。
  • waitStatus：等待状态，可能取值有CANCELLED（1）、SIGNAL（-1）、CONDITION（-2）、PROPAGATE（-3）。
  • prev：前继节点
  • next：后继节点。
  • nextWaiter：条件队列，本文不做讨论。

需要注意的是，CLH队列中的头节点head代表的是已经获取到锁的线程，其无需再关联线程thread。

图解并发编程-AQS-CLH.drawio

2.3. CLH控制流程

  • 入队流程：当线程获取互斥量失败时，会加入CLH等待队列中去，如果没有头节点则需要先创建头节点。
  • 挂起流程：加入到CLH等待队列中的线程并不会直接挂起，总的来看有三次获取锁的机会：
    • 第一次：首次获取互斥量成功时。
    • 第二次：第一次失败后，线程加入到CLH等待队列后，前继节点为头节点且获取互斥量成功时。
    • 第三次：第二次失败后，修改前继节点的等待状态waitStatus为SIGNAL后，前继节点为头节点且获取互斥量成功时。
  • 唤醒流程：持有锁的线程会释放互斥量，并且唤醒头节点的后继节点关联的线程，该线程会重新尝试获取锁。
  • 出队流程：当头节点的后继节点替换掉头节点时，原头节点出队。

3. AbstractQueuedSynchronizer源码解析

3.1. acquire加锁过程

3.1.1. 尝试获取互斥量（tryAcquire）

首先，尝试获取互斥量tryAcquire，该方法由AbstractQueuedSynchronizer的子类实现。

  • 获取互斥量成功代表加锁成功，则直接返回。

  • 获取互斥量失败代表加锁失败，则需要将当前线程加入到等待队列中去。

如下代码：

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

3.1.2. 线程加入等待队列（addWaiter）

当获取互斥量失败时，需要将当前线程加入到等待队列中去。

  1. 创建当前线程对应的节点Node。
  2. 如果有尾节点tail，则直接将当前线程对应的节点插入到等待队列的尾部并返回。
  3. 如果没有头节点head，则先创建头节点。
  4. 将当前线程对应的节点插入到等待队列的尾部tail。

如下代码：

private Node addWaiter(Node mode) {
   	// 1. 创建当前线程对应的节点
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    // 2. 如果有尾节点，则直接将当前线程对应的节点插入到等待队列的尾部
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

private Node enq(final Node node) {
    // CAS + 自旋
    for (;;) {
        Node t = tail;
        // 3. 如果没有头节点，则先创建头节点
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 4. 将当前线程对应的节点插入到等待队列的尾部
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

3.1.3. 重新尝试获取互斥量（acquireQueued）

  • 如果当前节点的前继节点是头节点且获取互斥量成功，则切换头节点为当前节点。
  • 若上一步失败，则判断是否需要挂起当前节点关联的线程，避免无限自旋浪费CPU资源。
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // CAS + 自旋
        for (;;) {
            final Node p = node.predecessor();
            // 如果当前节点的前继节点是头节点且获取互斥量成功
            if (p == head && tryAcquire(arg)) {
                // 切换头节点为当前节点
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 若上一步失败，则判断是否需要挂起当前节点关联的线程，避免无限自旋浪费CPU资源
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

3.1.4. 挂起线程（parkAndCheckInterrupt）

  • 如果当前节点的前继节点的等待状态waitStatus是0，则修改前继节点的状态为SIGNAL并返回false，意味着允许下次自旋，暂时不挂起当前节点关联的线程。
  • 如果当前节点的前继节点的等待状态waitStatus是SIGNAL，则返回true，意味着需要挂起当前节点关联的线程。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
	// 当前节点的前继节点的等待状态是 SIGNAL，意味着需要挂起当前节点关联的线程。
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        // 跳过被取消的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        // 当前节点的前继节点的等待状态是0，则修改前继节点的状态为 SIGNAL，意味着允许下次自旋，暂时不挂起当前节点关联的线程
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}


private final boolean parkAndCheckInterrupt() {
    // 挂起当前线程
    LockSupport.park(this);
    return Thread.interrupted();
}

3.2. release释放锁过程

3.2.1. 尝试释放互斥量（tryRelease）

尝试释放互斥量（tryRelease），该方法由AbstractQueuedSynchronizer的子类实现。

  • 释放互斥量成功代表释放锁成功，则唤醒后继节点并返回true。
  • 释放互斥量失败代表释放锁失败，则返回false。
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        // 如果当前节点的等待状态不是0的话，则唤醒后继节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

3.2.2. 唤醒后继节点（unparkSuccessor）

如果当前节点的状态是 SIGNAL，则切换当前节点的等待状态为0，再唤醒当前节点的后继节点关联的线程。

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.
     */
    // 当前节点的状态是 SIGNAL，则切换当前节点的等待状态为0
    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 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);
}

3.3. cancelAcquire取消加锁过程

  1. 将当前节点关联的线程设置为空；
  2. 跳过取消的前继节点；
  3. 将当前节点的等待状态设置为CANCELLED；
  4. 如果当前节点是尾节点，则移除该节点；否则如果当前节点的前继节点有能力唤醒当前节点的后继节点，则直接将前继节点指向后继节点；否则需要唤醒当前节点。
private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    // 将当前节点关联的线程设置为空
    node.thread = null;

    // Skip cancelled predecessors
    // 跳过取消的前继节点
    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.
    // 将当前节点的等待状态设置为 CANCELLED
    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
    }
}

4. ReentrantLock源码解析

从上面的类图可以看到ReentrantLock使用Sync完成锁的添加和释放，有两种锁分别是公平锁FairSync和非公平锁NonfairSync。

4.1. 公平锁FairSync的加锁

  • 如果等待队列中没有在等待的线程且设置互斥量state成功，则获取锁成功。
  • 如果当前线程已经持有锁，则递增互斥量state以重入锁。
  • 否则获取锁失败。
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        acquire(1);
    }

    /**
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        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;
    }
}

4.2. 非公平锁NofairSync的加锁

  • 如果设置互斥量state成功，则获取锁成功。
  • 如果当前线程已经持有锁，则递增互斥量state。
  • 否则获取锁失败。
abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        /**
         * Performs {@link Lock#lock}. The main reason for subclassing
         * is to allow fast path for nonfair version.
         */
        abstract void lock();

        /**
         * Performs non-fair tryLock.  tryAcquire is implemented in
         * subclasses, but both need nonfair try for trylock method.
         */
        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;
        }
        
        // ...
}
/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

4.3. 释放锁

  • 如果当前线程不是持有锁的线程，则抛出IllegalMonitorStateException异常。
  • 若果当前线程是持有锁的线程，则递减互斥量state。
  • 如果互斥量state为0，则释放锁成功。
abstract static class Sync extends AbstractQueuedSynchronizer {
        // ...
        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
        // ...
}
上一篇：CAS和原子类是如何实现的？ 下一篇：ThreadPoolExecutor线程池是如何实现的？

目录

  1. 1. 背景
  2. 2. AbstractQueuedSynchronizer设计思想
    1. 2.1. 互斥量的维护
    2. 2.2. CLH队列结构
    3. 2.3. CLH控制流程
  3. 3. AbstractQueuedSynchronizer源码解析
    1. 3.1. acquire加锁过程
      1. 3.1.1. 尝试获取互斥量（tryAcquire）
      2. 3.1.2. 线程加入等待队列（addWaiter）
      3. 3.1.3. 重新尝试获取互斥量（acquireQueued）
      4. 3.1.4. 挂起线程（parkAndCheckInterrupt）
    2. 3.2. release释放锁过程
      1. 3.2.1. 尝试释放互斥量（tryRelease）
      2. 3.2.2. 唤醒后继节点（unparkSuccessor）
    3. 3.3. cancelAcquire取消加锁过程
  4. 4. ReentrantLock源码解析
    1. 4.1. 公平锁FairSync的加锁
    2. 4.2. 非公平锁NofairSync的加锁
    3. 4.3. 释放锁