20230608 java.util.concurrent.locks.AbstractQueuedSynchronizer

介绍

• java.util.concurrent.locks.AbstractQueuedSynchronizer
• public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable
• 提供一个框架，用于实现依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关同步器（信号量、事件等）
• 此类旨在成为大多数依赖单个原子int值来表示状态的同步器的有用基础。子类必须定义更改此状态的受保护方法，并根据获取或释放此对象来定义该状态的含义。鉴于这些，此类中的其他方法执行所有排队和阻塞机制。子类可以维护其他状态字段，但只有使用方法 getState 、 setState 和 compareAndSetState 操作的原子更新的 int 值在同步方面被跟踪。
• 子类应定义为非公共内部帮助程序类，用于实现其封闭类的同步属性，参考 java.util.concurrent.locks.ReentrantLock.Sync
• 此类支持默认独占模式和共享模式，通常，实现子类仅支持这些模式中的一种，但两者都可以发挥作用，例如在ReadWriteLock中。仅支持独占或共享模式的子类不需要定义支持未使用模式的方法。
• 这个类定义了一个嵌套的AbstractQueuedSynchronizer.ConditionObject类，它可以被支持独占模式的子类用作Condition实现
• 方法isHeldExclusively报告同步是否相对于当前线程独占地保持
• 用当前getState值调用的方法release完全释放这个object
• 用当前getState值调用的方法acquire ，最终将这个对象恢复到它以前获得的状态
• 要使用此类作为同步器的基础，请重新定义以下方法（如果适用），方法是使用 getState 、 setState 或 compareAndSetState 检查或修改同步状态：
  • tryAcquire
  • tryRelease
  • tryAcquireShared
  • tryReleaseShared
  • isHeldExclusively
  • 这些方法的实现必须是内部线程安全的，并且通常应该很短并且不会阻塞。定义这些方法是唯一受支持的使用此类的方法。所有其他方法都声明为final方法，因为它们不能独立变化。
• 即使此类基于内部 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;
  

• 因为在入队之前调用 acquire 中的检查，所以新获取的线程可能会抢在其他被阻塞和排队的线程之前。但是，如果需要，您可以定义 tryAcquire 和/或 tryAcquireShared 以通过在内部调用一种或多种检查方法来禁用闯入，从而提供公平的FIFO 获取顺序。
• 特别是，大多数公平同步器可以定义tryAcquire以在hasQueuedPredecessors （一种专门为公平同步器使用的方法）返回false时返回true
• 默认 barging（也称为greedy 、 renouncement和convoy-avoidance ）策略的吞吐量和可扩展性通常最高。虽然这不能保证公平或无饥饿，但允许较早排队的线程在较晚排队的线程之前重新竞争，并且每次重新竞争都有机会成功对抗传入线程。
• 此类通过将其使用范围专门用于可以依赖int状态、获取和释放参数以及内部 FIFO 等待队列的同步器，部分地为同步提供了高效且可扩展的基础。当这还不够时，您可以使用atomic类、您自己的自定义java.util.Queue类和LockSupport阻塞支持从较低级别构建同步器。

API

protected 方法

protected final

state

• getState
• setState
• compareAndSetState

acquire/release

• tryAcquire

  • 独占锁需要实现
  • 尝试以独占模式获取
  • 此方法应查询对象的状态是否允许以独占模式获取它，如果允许则获取它
  • 可用于实现方法 Lock.tryLock()

• tryRelease

  • 独占锁需要实现
  • 尝试设置 state 以反映独占模式下的释放

• tryAcquireShared

  • 共享锁需要实现
  • 尝试以共享模式获取
  • 此方法应查询对象的状态是否允许以共享模式获取它，如果允许则获取它

• tryReleaseShared

  • 共享锁需要实现
  • 尝试设置 state 以反映共享模式下的释放

• isHeldExclusively

  • 判断是否当前线程独占
  • 需要子类实现

public 方法

acquire/release

• acquire

  • final void acquire(int arg)
  • 以独占模式获取
  • 忽略中断
  • 通过至少调用一次 tryAcquire 并在成功时返回来实现。否则线程会排队，可能会重复阻塞和解除阻塞，调用 tryAcquire 直到成功
  • 此方法可用于实现方法 Lock.lock

• acquireInterruptibly

  • final void acquireInterruptibly(int arg) throws InterruptedException
  • 以独占模式获取
  • 支持中断
  • 通过首先检查中断状态，然后至少调用一次tryAcquire ，成功返回来实现。否则线程将排队，可能会重复阻塞和解除阻塞，调用tryAcquire直到成功或线程被中断
  • 此方法可用于实现方法 Lock.lockInterruptibly

• tryAcquireNanos

  • final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException
  • 尝试以独占模式获取
  • 支持中断
  • 超时返回 false
  • 通过首先检查中断状态，然后至少调用一次tryAcquire ，成功返回来实现。否则，线程将排队，可能会重复阻塞和解除阻塞，调用tryAcquire直到成功或线程被中断或超时结束
  • 此方法可用于实现方法 Lock.tryLock(long, TimeUnit)

• release

  • final boolean release(int arg)
  • 以独占模式释放
  • 如果tryRelease返回 true，则通过取消阻塞一个或多个线程来实现
  • 此方法可用于实现方法 Lock.unlock

• acquireShared

  • 共享模式
  • 参考 acquire

• acquireSharedInterruptibly

  • 共享模式
  • 参考 acquireInterruptibly

• tryAcquireSharedNanos

  • 共享模式
  • 参考 tryAcquireNanos

• releaseShared

  • 共享模式
  • 参考 release

queue

• hasQueuedThreads
  • 查询是否有任何线程正在等待获取
• hasContended
  • 查询是否有线程争用过这个同步器
  • 也就是说，如果 acquire 方法曾经被阻塞
• getFirstQueuedThread
  • 返回队列中的第一个（等待时间最长的）线程，如果当前没有线程排队，则返回null
• isQueued
  • 如果给定线程当前正在排队，则返回 true
• hasQueuedPredecessors
  • 查询是否有任何线程等待获取的时间比当前线程长
• getQueueLength
  • 返回等待获取的线程数的估计值
• getQueuedThreads
  • 返回一个集合，其中包含可能正在等待获取的线程
• getExclusiveQueuedThreads
  • 返回一个集合，其中包含可能正在等待以独占模式获取的线程
  • 这与 getQueuedThreads 具有相同的属性，只是它只返回那些因独占获取而等待的线程
• getSharedQueuedThreads
  • 返回一个集合，其中包含可能正在等待以共享模式获取的线程
  • 这与 getQueuedThreads 具有相同的属性，除了它只返回那些由于共享获取而等待的线程

Condition

• owns
  • final boolean owns(ConditionObject condition)
  • 查询入参 ConditionObject 是否使用此同步器作为其锁
• hasWaiters
  • final boolean hasWaiters(ConditionObject condition)
  • 查询是否有任何线程正在等待与此同步器关联的 Condition
• getWaitQueueLength
  • final int getWaitQueueLength(ConditionObject condition)
  • 返回等待与此同步器关联的 Condition 的线程数的估计值
• getWaitingThreads
  • final Collection<Thread> getWaitingThreads(ConditionObject condition)
  • 返回一个集合，其中包含可能正在等待与此同步器关联的 Condition 的那些线程

继承父类 java.util.concurrent.locks.AbstractOwnableSynchronizer

• setExclusiveOwnerThread , getExclusiveOwnerThread

其他

• toString

代码理解

state

定义 volatile 字段 state

private volatile int state;

protected final int getState() {
    return state;
}

protected final void setState(int newState) {
    state = newState;
}

独占锁和共享锁

通过方法 isHeldExclusively 判断锁是否独占

• 独占，重写方法 tryAcquire , tryRelease
  • 参考 ReentrantLock
• 共享，重写方法 tryAcquireShared , tryReleaseShared
  • 参考 ReentrantReadWriteLock

Node

private transient volatile Node head;

private transient volatile Node tail;

abstract static class Node {
    volatile Node prev;
    volatile Node next;
    Thread waiter;
    volatile int status;
}

Node 在 AQS 内部有三个子类：

• ExclusiveNode
• SharedNode
• ConditionNode

acquire 方法

java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire(java.util.concurrent.locks.AbstractQueuedSynchronizer.Node, int, boolean, boolean, boolean, long)

入参：

• Node node
  • 除非重新获取 Condition ，否则为空
• int arg
  • 获取参数
• boolean shared
  • 是否共享锁模式
• boolean interruptible
  • 是否支持中断
• boolean timed
  • 是否支持超时
• long time
  • 超时时长，单位 ns

返回值：如果获取成功则为正，如果超时则为 0，如果中断则为负

方法的主要逻辑是一个无限循环，直到获取成功、超时或中断。在循环中，它会执行以下操作：

1. 检查节点是否是第一个节点。如果是，则确保头节点稳定；否则，确保有效的前驱节点。
2. 如果节点是第一个节点或尚未入队，尝试获取。
3. 如果节点尚未创建，创建它。
4. 如果尚未入队，尝试入队一次。
5. 如果从 park 中唤醒，重试（最多 postSpins 次）。
6. 如果 WAITING 状态未设置，设置并重试。
7. 否则，park 并清除 WAITING 状态，并检查取消。

持有锁的线程不在队列中

队列的第一个 Node 的 thread 是 null

park 阻塞线程

LockSupport.park(this);

然后在 release 方法中 unpark 后，继续执行 acquire 方法的下一个语句

return cancelAcquire(node, interrupted, interruptible);

在 cancelAcquire 方法中对 Node 队列进行操作

这里不是在 release 方法中对 Node 队列进行操作，感觉有点反常识，确实精彩