基于AbstractQueuedSynchronizer之Condition源码分析-Toy模板网

这篇具有很好参考价值的文章主要介绍了基于AbstractQueuedSynchronizer之Condition源码分析。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

java.util.concurrent.locks.Condition 是一个将 Object 的监视器方法（wait、notify 和 notifyAll）分离到不同的对象中的类，以实现每个对象有多个等待集合的效果，这些对象与任意 Lock 实现结合使用。当一个 Lock 替换了synchronized方法和语句的使用时，Condition 替换了 Object 监视器方法的使用。

Conditions（也称为条件队列或条件变量）提供了一种方法，使一个线程可以在等待被另一个线程通知某个状态条件可能已经变为真之前，挂起执行。由于对这个共享状态信息的访问发生在不同的线程中，因此必须保护它，因此某种形式的锁与条件相关联。等待条件提供的主要属性是它原子性地释放相关联的锁并挂起当前线程，就像 Object.wait 一样。

Condition 实例与锁紧密关联。要为特定的 Lock 实例获得一个 Condition实例，请使用其 Lock.newCondition() 方法。

一个Condition的例子(生产者-消费者)，假设我们有一个支持 put 和 take 方法的有界缓冲区。如果尝试在空缓冲区上执行 take，则线程将阻塞，直到有一个项可用；如果尝试在满缓冲区上执行 put，则线程将阻塞，直到有一个空间可用。我们希望将等待 put 线程和等待 take 线程保持在不同的等待集中，以便在缓冲区中可用项或空间变得可用时，我们可以使用一次只通知一个线程的优化。这可以通过使用两个 Condition 实例实现。

class BoundedBuffer<E> {
    final Lock lock = new ReentrantLock();
    final Condition notFull  = lock.newCondition(); 
    final Condition notEmpty = lock.newCondition(); 
 
    final Object[] items = new Object[100];
    int putptr, takeptr, count;
 
    public void put(E x) throws InterruptedException {
      lock.lock();
      try {
        while (count == items.length) // 使用while，避免虚假唤醒
          notFull.await();
        items[putptr] = x;
        if (++putptr == items.length) putptr = 0;
        ++count;
        notEmpty.signal();
      } finally {
        lock.unlock();
      }
    }
 
    public E take() throws InterruptedException {
      lock.lock();
      try {
        while (count == 0)
          notEmpty.await();
        E x = (E) items[takeptr];
        if (++takeptr == items.length) takeptr = 0;
        --count;
        notFull.signal();
        return x;
      } finally {
        lock.unlock();
      }
    }
  }

Condition 实现可以提供超出 Object监视器方法的行为和语义，例如保证排序、非可重入使用或超时。

Condition 接口提供了以下方法：

await()：导致当前线程等待，直到它被通知或中断。
awaitUninterruptibly()：导致当前线程等待，直到它被通知。
awaitNanos(long nanosTimeout)：导致当前线程等待，直到它被通知，或者指定的等待时间已经过去。
awaitUntil(Date deadline)：导致当前线程等待，直到它被通知，或者指定的最后期限已经过去。
signal()：唤醒一个等待在这个条件上的单个线程。
signalAll()：唤醒所有等待在这个条件上的线程。

Condition接口的实现可以提供与 Object 监视器方法不同的行为和语义，例如保证通知的顺序，或者在执行通知时不需要持有锁。如果实现提供了这样的专用语义，那么实现必须记录这些语义。

注意，Condition 实例只是普通对象，它们本身可以在同步语句中作为目标使用，并且可以调用它们自己的监视器等待和通知方法。获取 Condition 实例的监视器锁，或者使用它的监视器方法，与获取与该 Condition 相关联的锁或使用它的等待和信号方法之间没有指定的关系。为避免混淆，建议您永远不要以这种方式使用 Condition 实例，除非在它们自己的实现中。

除非另有说明，否则将任何参数传递为 null 值都将导致抛出 NullPointerException。

实现注意事项

在等待 Condition 时，**“虚假唤醒”**是允许发生的，一般来说，这是对底层平台语义的让步。这对大多数应用程序程序几乎没有实际影响，因为应该总是在循环中等待 Condition，并测试正在等待的状态谓词。实现可以自由地消除虚假唤醒的可能性，但建议应用程序员总是假定它们可能会发生，因此总是在循环中等待。

条件等待的三种形式（可中断、不可中断和定时）在一些平台上的实现难度和性能特征可能不同。特别地，难以提供这些功能并保持特定的语义，例如保证排序保证。此外，对线程的实际挂起进行中断的能力可能在所有平台上都无法实现。

因此，实现不需要为所有三种等待形式定义完全相同的保证或语义，也不需要支持中断线程挂起的实际。实现需要清楚地记录每个等待方法提供的语义和保证，当实现支持线程挂起的中断时，它必须遵守此接口中定义的中断语义。

由于中断通常意味着取消，并且检查中断通常不频繁，因此实现可以优先响应中断而不是正常的方法返回。即使可以证明中断发生在可能解除线程阻塞的其他操作之后，这也是正确的。实现应记录此行为。

ConditionObject

AQS(AbstractQueuedSynchronizer)中的ConditionObject类实现了Condition接口。

  public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        // transient不参与序列化的意思，
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;

        /**
         * 构造函数
         */
        public ConditionObject() { }
      //省略后续代码................

然后我们再回顾一下Node的属性：

prev和next用于实现阻塞队列的双向链表，这里的nextWaiter用于实现条件队列的单向链表

volatile int waitStatus; // 可取值 0、CANCELLED(1)、SIGNAL(-1)、CONDITION(-2)、PROPAGATE(-3)
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;

接下来，我们按照流程来分析源码，先查看await方法：

		/**
         * 实现可中断的条件等待。
         * 1、如果当前线程被中断，则抛出InterruptedException异常
         * 2、保存getState返回的锁状态
         * 3、调用release并将保存的状态作为参数传递，如果release失败则抛出异常
         *    throwing IllegalMonitorStateException
         * 4、阻塞直到被唤醒或中断
         * 5、通过调用带有保存的锁状态作为参数的专用版本的acquire来重新获取锁
         * 6、如果在步骤4中被阻塞中断，将抛出InterruptedException异常
         */
public final void await() throws InterruptedException {
            // 既然方法要响应中断，那么在最开始就判断中断状态
    		// 不可被中断的是另一个方法awaitUninterruptibly()
            if (Thread.interrupted())
                throw new InterruptedException();
    		// 添加到condition的条件队列中
            Node node = addConditionWaiter();
    		// 释放锁，返回值就是getState返回的锁状态
            int savedState = fullyRelease(node);
            int interruptMode = 0;
    		// while循环，退出条件有两种
    		// 1.isOnSyncQueue返回true
    		// 2. checkInterruptWhileWaiting(node)) != 0,代表线程中断
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
    		// 被唤醒后，将进入阻塞队列，等待获取锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

1、将节点加入到条件队列

addConditionWaiter并不是绝对的线程安全的。首先条件队列中出现多个Node节点的情况，我认为是出现在获取到锁之后在锁的范围内部使用多线程同时调用await方法，这种情况下它们的Node节点被添加到了同一个队列中。在这种情况下，由于多个线程在同一个时刻访问条件队列，出现竞争条件，此时源代码中是无法保证线程安全的，需要我们自己去处理来保证线程安全。

// Adds a new waiter to wait queue.
private Node addConditionWaiter() {
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) {
                // 这个方法会遍历整个条件队列，将已取消的所有节点清理掉
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
    		// node节点初始化
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

上面的代码还是比较简单，就是将node节点放入队尾。

在addConditionWaiter方法中，有一个unlinkCancelledWaiters方法，该方法用于清理条件队列中已取消的等待节点。在这里我不是很理解为什么最后一个等待节点不是等待状态(Node.CONDITION)才去清理条件队列。

 private void unlinkCancelledWaiters() {
            Node t = firstWaiter;
     		// 重生之我是新队列
            Node trail = null;
            while (t != null) {
                // 拿到下一个节点，用来后续判断和下一次while需要使用的节点
                Node next = t.nextWaiter;
                // 如果当前节点的状态不是CONDITION状态，也就是其他需要被取消的状态
                if (t.waitStatus != Node.CONDITION) {
                    // 当前节点和后续断开
                    t.nextWaiter = null;
                    // 第一个就是取消的节点，那么改变firstWaiter
                    if (trail == null)
                        firstWaiter = next;
                    // 否则把结点连接起来
                    else
                        trail.nextWaiter = next;
                    // 为空的时候把trail给lastWaiter
                    if (next == null)
                        lastWaiter = trail;
                }
                else
                    trail = t;
                t = next;
            }
        }

2、完全释放独占锁

回到await方法，节点入队了以后，会调用int savedState = fullyRelease(node)释放锁并返回锁的状态值，这里是完全释放独占锁，因为ReentrantLock可重入的。

final int fullyRelease(Node node) {
    	// 标记，是否给当前node节点标识为CANCELLED状态
        boolean failed = true;
        try {
            // 获取锁的state
            int savedState = getState();
            // 释放锁
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

如果一个线程并没有持有锁，但是它执行了condition.await方法，Doug Lea其实并没有不允许这么做，而是允许它进入到条件队列中，但是上面的这个方法，由于它并不持有锁，那么release这个方法是姚返回false的，并且走入到异常分支抛出异常。这样在finally中将这个节点标记为CANCELLED，后续就会被清理出去了。

3、等待进入阻塞队列

isOnSyncQueue用于判断node节点是否已经转移到同步队列(阻塞队列)。

final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
        return findNodeFromTail(node);
    }

4、signal唤醒线程，转移到同步队列

LockSupport.park(this)把线程挂起了之后需要等待唤醒。

唤醒操作通常由另一个线程来操作，就像生产者-消费者模式中，如果线程因为等待消费而挂起，那么当生产者生产一个东西后，会调用signal唤醒正在等待的线程来消费。

// 唤醒等待了最久的线程
// 其实就是将这个线程对应的node从条件队列移动到同步队列
public final void signal() {
    // 调用signal方法的线程必须持有当前的独占锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 让first节点独立，断开关联，方便执行后续操作
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
    	// CAS去修改waitStatus,如果不能修改成功，这个node节点已经被取消了
        // 方法直接返回false，继续执行下一个节点。
    	// CAS修改成功会设置waitStatus为0
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

        /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */
    	// 自旋入队，返回的是同步队列的前驱节点
        Node p = enq(node);
    	// ws>0 说明node在同步队列重的前驱节点取消了等待锁，直接唤醒对应的线程
    	// ws<=0 那么CAS将p的状态设置为SIGNAL，表明后继节点需要被唤醒
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

5、唤醒后检查中断状态

执行signal之后，我们的线程由条件队列转移到了同步队列，也就是说在同步队列中我们的线程节点被唤醒之后，才会继续往下执行。

int interruptMode = 0;
while (!isOnSyncQueue(node)) {
    // 线程挂起
    LockSupport.park(this);
	
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
}

查看checkInterruptWhileWaiting方法可以知道interruptMode的值分为0，1，-1。

// Mode meaning to reinterrupt on exit from wait
private static final int REINTERRUPT =  1;

// Mode meaning to throw InterruptedException on exit from wait
private static final int THROW_IE    = -1;

// or 0 if not interrupted.
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :0;
}

REINTERRUPT：代表 await 返回的时候，需要重新设置中断。
THROW_IE：代表 await 返回的时候，需要抛出InterruptedException 异常。
0：说明在 await 期间，没法发生中断。

有以下四种情况会让 LockSupport.park(this); 这句返回继续往下执行：

常规路径。signal -> 转移节点到阻塞队列 -> 获取了锁（unpark）
线程中断。在 park 的时候，另外一个线程对这个线程进行了中断
signal 的时候我们说过，转移以后的前驱节点取消了，或者对前驱节点的CAS操作失败了
假唤醒。这个也是存在的，和 Object.wait() 类似，都有这个问题

插播一些小知识:

线程中断可以唤醒通过LockSupport.park()方法休眠的线程。当一个线程被中断时，它会从park()方法调用中返回，并且不会阻塞在此方法上。线程的中断状态将被设置为true，可以通过调用Thread.interrupted()方法来检查线程是否被中断。

需要注意的是，如果线程在调用park()方法之前已经被中断，则park()方法会立即返回而不会阻塞线程，并且线程的中断状态也会被设置为true。此外，LockSupport.park()方法还可以被其他因素唤醒，例如调用LockSupport.unpark(Thread)方法，或者当一个线程获取一个被另一个线程持有的锁时。

假唤醒（Spurious wakeup）是指在多线程编程中，一个线程在没有被唤醒的情况下，从等待状态（如wait()或LockSupport.park()等方法）中返回。这种情况可能会发生在某些操作系统上，尤其是在多处理器系统上。

假唤醒通常是由操作系统或硬件中的某些因素引起的，而不是由程序中的代码引起的。例如，在某些操作系统中，当一个进程接收到信号时，所有等待该进程的线程都会被唤醒，即使信号与这些线程无关。在这种情况下，被唤醒的线程可能会发现它们并没有实际上需要的条件满足，因此它们会重新进入等待状态。

为了防止假唤醒，通常需要在程序中使用循环来检查等待条件是否满足，而不仅仅是在wait()或LockSupport.park()方法返回时假设条件已经满足。例如，在使用wait()方法等待一个条件时，通常应该在一个循环中调用wait()方法，如下所示：

synchronized (lock) {
    while (!condition) {
        lock.wait();
    }
    // do something when condition is true
}

这样，在假唤醒发生时，线程将重新检查等待条件是否满足，并在条件满足时继续执行，而不是在条件未满足时错误地执行下一步操作。

ok，继续往下走查看checkInterruptWhileWaiting方法

线程被唤醒后的第一件事情就是执行checkInterruptWhileWaiting方法，次方法用于判断是否在线程挂起期间发生了中断，如果没有则返回0。如果发生了中断则执行transferAfterCancelledWait方法。

当一个节点被唤醒时,它可能处于条件队列,也可能处于同步队列中.如果是出入条件队列中,那么需要被转移到同步队列

//Transfers node, if necessary, to sync queue after a cancelled wait. 
//Returns true if thread was cancelled before being signalled.
// 只有线程处于中断才会调用次方法
final boolean transferAfterCancelledWait(Node node) {
        // CAS将waitStatus设置为0
    	// 如果设置成功，说明是在signal方法调用之前发生的中断，因为如果是signal方法先调用，此处waitStatus就是0了。
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            // 将node节点转移到同步队列并放回true
            // 也就是说即使发生了中断，还是会入队.
            enq(node);
            return true;
        }
        /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
        while (!isOnSyncQueue(node))
            // 要么node在同步队列,要么中断才会退出while
            Thread.yield();
        // CAS失败,也就是在signal方法调用后出现的中断
        return false;
    }

这个while什么作用?

也就是signal唤醒的时候node节点还没有入同步队列,但是线程被唤醒了,所以此处就硬等。

6、获取独占锁

await方法的while循环出来之后，就是下面这段代码：

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
    unlinkCancelledWaiters();
if (interruptMode != 0)
    reportInterruptAfterWait(interruptMode);

acquireQueued方法有两个参数，一个node一个savedState。因为是可重入锁，之前await释放了多少锁就得加回来。这里就不细说获取锁得方法了。这里主要是AQS得内容。

7、处理中断状态

到这里，我们终于知道interruptMode是干嘛的了

REINTERRUPT：重新中断当前线程，因为它代表 await() 期间没有被中断，而是 signal() 以后发生的中断。
THROW_IE：await 方法抛出 InterruptedException 异常，因为它代表在 await() 期间发生了中断。
0：说明在 await 期间，没法发生中断。

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

*带超时机制的和不抛出InterruptedException的awati

带超时机制的awiat如下，源码相对简单

public final long awaitNanos(long nanosTimeout) 
                  throws InterruptedException
public final boolean awaitUntil(Date deadline)
                throws InterruptedException
public final boolean await(long time, TimeUnit unit)
                throws InterruptedException

不抛出InterruptedException的await：awaitUninterruptibly文章来源地址https://www.toymoban.com/news/detail-490137.html

public final void awaitUninterruptibly() {
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    boolean interrupted = false;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if (Thread.interrupted())
            interrupted = true;
    }
    if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();
}