从ReentrantLock看AQS
大黄 Lv4
  2022-03-19 18:36:13 2022-03-19 18:36    3k Words  14 Mins  

从ReentrantLock看AQS

AQS的三个核心点

  • state
  • 协作类实现的获取锁/释放锁的方法
  • FIFO队列

关于state

state是用来判断是否有线程占用当前锁，与另一个参数exclusiveOwnerThread 配合使用

以ReentrantLock获取锁为例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
 * ReentrantLock 获取非公平锁的代码
 */
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() {
				// 设置state => 从0到1 [即占有锁]
        if (compareAndSetState(0, 1))
						// 成功 => 将占用的线程改成当前线程
            setExclusiveOwnerThread(Thread.currentThread());
        else
						// 设置失败，说明当前的锁被其他线程占用，尝试获取锁 [见下一段代码]
            acquire(1);
    }

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

关于协作类实现的释放锁/获取锁的方法

以上面的例子为例，当线程要获取的锁被其他线程占用的时候，就需要我们去自定一个获取锁的逻辑

1
2
3
4
5
6
7
public final void acquire(int arg) {
		// tryAcquire 就是协作类自定义的获取锁的逻辑
    if (!tryAcquire(arg) &&
				// 获取失败，统一交给AQS管理（添加等待节点，放入队列中，将当前线程挂起）-这套属于固有的逻辑，不需要协作类去实现（实现成本高，且属于重复代码）
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

那么来重点看看 tryAcquire 方法 （接着以非公平锁为例）

1
2
3
protected final boolean tryAcquire(int acquires) {
  return nonfairTryAcquire(acquires);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
		// 再次获取一下state
    int c = getState();
    if (c == 0) {
				// 说明锁被释放，再次cas设置state
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
		// 如果持有锁的线程是当前线索，state+1（可重入）
    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;
}

再次尝试，获取的到锁还好说。假如获取不到锁，就需要用到了刚刚提到的FIFO队列

AQS核心内容 FIFO队列及入队出队规则-入队

此处结合着我们刚刚将的流程来，不单独针对各个点做叙述

单独再将这块代码拿出来

1
2
3
4
if (!tryAcquire(arg) &&
		// 获取失败，统一交给AQS管理（添加等待节点，放入队列中，将当前线程挂起）-这套属于固有的逻辑，不需要协作类去实现（实现成本高，且属于重复代码）
    acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 具体见下段代码
    selfInterrupt();

添加等待节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
private Node addWaiter(Node mode) {
		// 新建一个节点对象
    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;
				// CAS将当前节点设置为尾节点[链表添加尾节点]
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
		// 入队（此处是没有等待节点的情况）
    enq(node);
    return node;
}
// 关于上面代码的一段思考：如果同时多个线程进入，会不会有并发问题？
// ===> 不会，有一个线程会cas操作成功设置，其它cas失败的线程会执行enq的方法入队

// CAS操作
private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

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

尝试让队列中的头节点获取锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
						// 获取当前节点的前置节点
            final Node p = node.predecessor();
						// 前置节点是head，说明当前节点是第一个等待锁的节点（此时也会让当前节点再次去尝试获取锁 即tryAcquire方法）
            if (p == head && tryAcquire(arg)) {
								// 获取锁成功的处理
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
						// 获取锁失败的处理 shouldParkAfterFailedAcquire （这个方法一会单独讲）
            if (shouldParkAfterFailedAcquire(p, node) &&
										// 挂起当前线程
                		parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
				// 响应中断
        if (failed)
						// 撤销尝试获取锁（这个也一会在再讲）
            cancelAcquire(node);
    }
}

是否要挂起当获取锁失败方法解析 shouldParkAfterFailedAcquire

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
		// 前缀节点的等待状态
	  int ws = pred.waitStatus;
	  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;
						// >0 说明前置节点已经被取消，可以直接删除
	      } 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.
	       */
				// 将前置节点设置为SIGNAL的状态，先入队的线程先被唤醒
	      compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
	  }
	  return false;
}

此处放一个WaitStatus枚举的状态表

1
2
3
4
5
6
7
8
9
10
11
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED =  1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL    = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
 * waitStatus value to indicate the next acquireShared should
 * unconditionally propagate
 */
static final int PROPAGATE = -3;

WaitStatus枚举含义

SATUS 当一个Node被初始化的时候的默认值
CANCELLED 为1，表示线程获取锁的请求已经取消了
CONDITION 为-2，表示节点在等待队列中，节点线程等待唤醒
PROPAGATE 为-3，当前线程处在SHARED情况下，该字段才会使用
SIGNAL 为-1，表示线程已经准备好了，就等资源释放了

释放锁的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void unlock() {
    sync.release(1);
}

public final boolean release(int arg) {
		// 尝试释放锁
    if (tryRelease(arg)) {
        Node h = head;
				// 如果有等待节点，尝试唤醒（即使创建的临时节点，也会在实际入队的过程中将临时节点改成SIGNAL状态）
        if (h != null && h.waitStatus != 0)
						// 唤醒
            unparkSuccessor(h);
        return true;
    }
    return false;
}

tryRelease 释放锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
protected final boolean tryRelease(int releases) {
		// state - release 重入次数 - 释放次数
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread()) // 当前线程不为持有锁的线程
        throw new IllegalMonitorStateException();
    boolean free = false;
		// state = 0，代表释放锁
    if (c == 0) {
        free = true;
      	// 全部释放
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

AQS核心内容 FIFO队列及入队出队规则-出队

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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)
				// 头节点<0，设置为初始状态 （小于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)
				// 唤醒节点，再走入刚刚的acquireQueued方法
        LockSupport.unpark(s.thread);
}

加锁失败撤销等待节点 cancelAcquire

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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.
    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
    }
}

一个问题

关于刚刚为什么要从尾节点往前找去添加节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
// 尝试获取锁的入队操作
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  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 {
      // 删除cancel的节点(先建立后面指向前面的指针，再建立前面到后面的指针)
      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.
     */
    compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}

// 唤醒节点
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 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);
}

node.prev = pred; compareAndSetTail(pred, node) 这两个地方可以看作Tail入队的原子操作，但是此时pred.next = node;还没执行，如果这个时候执行了unparkSuccessor方法，就没办法从前往后找了，所以需要从后往前找。还有一点原因，在产生CANCELLED状态节点的时候，先断开的是Next指针，Prev指针并未断开，因此也是必须要从后往前遍历才能够遍历完全部的Node。

流程图

获取锁的流程图

ReentrantLock尝试获取锁.jpg

释放锁的流程图

ReentrantLock尝试释放锁.jpg

======================== 2023.11.24 更新 ========================

公平锁相关

之前讲的都是非公平锁，此处补充下公平锁，对比下两者之间的差距

1
2
3
4
// 尝试获取锁
final void lock() {
  acquire(1);
}
1
2
3
4
5
public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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;
}

// 逻辑相同
private Node addWaiter(Node mode) {
  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(node);
  return node;
}

// 逻辑相同
final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      final Node p = node.predecessor();
      // tryAcquire逻辑不同
      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);
  }
}

==> 结论：非公平锁和公平锁的差别就是在尝试获取锁的时候，公平锁会去看一下还有没有节点在等待获取锁，有的话等待的节点优先级高于当前节点。而非公平锁则是直接尝试获取锁，当前节点与其他等待节点进行竞争

  • Post title：从ReentrantLock看AQS
  • Post author：大黄
  • Create time：2022-03-19 18:36:13
  • Post link：https://huangbangjing.cn/2022/03/19/从ReentrantLock看AQS/
  • Copyright Notice：All articles in this blog are licensed under BY-NC-SA unless stating additionally.