gpt4 book ai didi

17.AQS中的Condition是什么?

转载 作者:我是一只小鸟 更新时间:2023-05-17 14:31:45 30 4
gpt4 key购买 nike

欢迎关注: 王有志 期待你加入Java人的提桶跑路群: 共同富裕的Java人 。

今天来和大家聊聊 Condition , Condition 为AQS“家族” 提供了等待与唤醒的能力 ,使AQS"家族"具备了像 synchronized 一样暂停与唤醒线程的能力。我们先来看两道关于 Condition 的面试题目:

  • Condition Object 的等待与唤醒有什么区别?
  • 什么是 Condition 队列?

接下来,我们就按照“是什么”,“怎么用”和“如何实现”的顺序来揭开 Condition 的面纱吧.

Condition是什么?

Condition 是Java中的接口,提供了与 Object#wait 和 Object#notify 相同的功能。Doug Lea在 Condition 接口的描述中提到了这点:

Conditions (also known as condition queues or condition variables) provide a means for one thread to suspend execution (to "wait") until notified by another thread that some state condition may now be true. 。

来看 Condition 接口中提供了哪些方法:

                        
                          public interface Condition {
  void await() throws InterruptedException;
  
  void awaitUninterruptibly();
  
  long awaitNanos(long nanosTimeout) throws InterruptedException;
  
  boolean await(long time, TimeUnit unit) throws InterruptedException;
  
  boolean awaitUntil(Date deadline) throws InterruptedException;
  
  void signal();
  
  void signalAll();
}

                        
                      

Condition 只提供了两个功能: 等待(await)和唤醒(signal) ,与 Object 提供的等待与唤醒时相似的:

                        
                          public final void wait() throws InterruptedException;
  
public final void wait(long timeoutMillis, int nanos) throws InterruptedException;

public final native void wait(long timeoutMillis) throws InterruptedException;

@HotSpotIntrinsicCandidate
public final native void notify();

@HotSpotIntrinsicCandidate
public final native void notifyAll();

                        
                      

唤醒功能上, Condition 与 Object 的差异并不大:

  • Condition#signal \(\approx\) Object#notify 。

  • Condition#signalAll \(=\) Object#notifyAll 。

多个线程处于等待状态时, Object#notify() 是“随机”唤醒线程,而 Condition#signal 则由具体实现决定如何唤醒线程,如: ConditionObject 唤醒的是最早进入等待的线程 , 但两个方法均只唤醒一个线程.

等待功能上, Condition 与 Object 的共同点是: 都会释放持有的资源 , Condition 释放锁 , Object 释放Monitor,即进入等待状态后允许其他线程获取锁/监视器 。主要的差异体现在 Condition 支持了更加丰富的场景,通过一张表格来对比下:

Condition 方法 Object 方法 解释
Condition#await() Object#wait() 暂停线程,抛出线程中断异常
Condition#awaitUninterruptibly() / 暂停线程,不抛出线程中断异常
Condition#await(time, unit) Object#wait(timeoutMillis, nanos) 暂停线程,直到被唤醒或等待指定时间后,超时后自动唤醒返回false,否则返回true
Condition#awaitUntil(deadline) / 暂停线程,直到被唤醒或到达指定时间点,超时后自动唤醒返回false,否则返回true
Condition#awaitNanos(nanosTimeout) / 暂停线程,直到被唤醒或等待指定时间后,返回值表示被唤醒时的剩余时间(nanosTimeout-耗时),结果为负数表示超时

除了以上差异外, Condition 还支持创建多个等待队列,即同一把锁拥有多个等待队列,线程在不同队列中等待,而 Object 只有一个等待队列 。《Java并发编程的艺术》中也有一张类似的表格,放在这里供大家参考:

Tips :

  • 实际上signal翻译为唤醒并不恰当~~
  • 涉及到 Condition 的实现部分,下文通过AQS中的 ConditionObject 详细解释。

Condition怎么用?

既然 Condition 与 Object 提供的等待与唤醒功能相同,那么它们的用法是不是也很相似呢?

与调用 Object#wait 和 Object#notifyAll 必须处于 synchronized 修饰的代码中一样(获取Monitor), 调用 Condition#await 和 Condition#signalAll 的前提是要先获取锁 。但不同的是,使用 Condition 前,需要先通过锁去创建 Condition .

以 ReentrantLock 中提供的 Condition 为例,首先是创建 Condition 对象:

                        
                          ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();

                        
                      

然后是获取锁并调用 await 方法:

                        
                          new Thread(() -> {
  lock.lock();
  try {
    condition.await();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  lock.unlock();
}

                        
                      

最后,通过调用 singalAll 唤醒全部阻塞中的线程:

                        
                          new Thread(() -> {
  lock.lock();
  condition.signalAll();
  lock.unlock();
}

                        
                      

ConditionObject的源码分析

作为接口 Condition 非常惨,因为在Java中只有AQS中的内部类 ConditionObject 实现了 Condition 接口:

                        
                          public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  
  public class ConditionObject implements Condition, java.io.Serializable {
    private transient Node firstWaiter;
    
    private transient Node lastWaiter;
  }
  
  static final class Node {
    // 省略
  }
}

                        
                      

ConditionObject 只有两个 Node 类型的字段,分别是链式结构中的头尾节点, ConditionObject 就是通过它们实现的等待队列。那么 ConditionObject 的等待队列起到了怎样的作用呢?是类似于AQS中的排队机制吗?带着这两个问题,我们正是开始源码的分析.

await方法的实现

Condition 接口中定义了4个线程等待的方法:

  • void await() throws InterruptedException
  • void awaitUninterruptibly();
  • long awaitNanos(long nanosTimeout) throws InterruptedException;
  • boolean await(long time, TimeUnit unit) throws InterruptedException;
  • boolean awaitUntil(Date deadline) throws InterruptedException;

方法虽然很多,但它们之间的差异较小,只体现在时间的处理上,我们看其中最常用的方法:

                        
                          public final void await() throws InterruptedException {
  // 线程中断,抛出异常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  // 注释1:加入到Condition的等待队列中
  Node node = addConditionWaiter();
  // 注释2:释放持有锁(调用AQS的release)
  int savedState = fullyRelease(node);
  int interruptMode = 0;
  // 注释3:判断是否在AQS的等待队列中
  while (!isOnSyncQueue(node)) {
    LockSupport.park(this);
    // 中断时退出方法
    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
      break;
    }
  }
  
  // 加入到AQS的等待队列中,调用AQS的acquireQueued方法
  if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
    interruptMode = REINTERRUPT;
  }
  
  // 断开与Condition队列的联系
  if (node.nextWaiter != null) {
    unlinkCancelledWaiters();
  }
  
  if (interruptMode != 0) {
   reportInterruptAfterWait(interruptMode);
  }
}

                        
                      

注释1的部分,调用 addConditionWaiter 方法添加到 Condition 队列中:

                        
                          private Node addConditionWaiter() {
  // 判断当前线程是否为持有锁的线程
  if (!isHeldExclusively()) {
    throw new IllegalMonitorStateException();
  }
  
  // 获取Condition队列的尾节点
  Node t = lastWaiter;
  // 断开不再位于Condition队列的节点
  if (t != null && t.waitStatus != Node.CONDITION) {
    unlinkCancelledWaiters();
    t = lastWaiter;
  }
  
  // 创建Node.CONDITION模式的Node节点
  Node node = new Node(Node.CONDITION);
  if (t == null) {
    // 队列为空的场景,将node设置为头节点
    firstWaiter = node;
  } else {
    // 队列不为空的场景,将node添加到尾节点的后继节点上
    t.nextWaiter = node;
  }
  // 更新尾节点
  lastWaiter = node;
  return node;
}

                        
                      

可以看到, Condition 的队列是一个朴实无华的双向链表,每次调用 addConditionWaiter 方法,都会加入到 Condition 队列的尾部.

注释2的部分, 释放线程持有的锁,同时移出AQS的队列 ,内部调用了AQS的 release 方法:

                        
                          =final int fullyRelease(Node node) {
  try {
    int savedState = getState();
    if (release(savedState)) {
      return savedState;
    }
    throw new IllegalMonitorStateException();
  } catch (Throwable t) {
    node.waitStatus = Node.CANCELLED;
    throw t;
  }
}

                        
                      

因为已经分析过AQS的 release 方法和 ReentrantLock 实现的 tryRelease 方法,这里我们就不过多赘述了.

注释3的部分, isOnSyncQueue 判断当前线程是否在AQS的等待队列中,我们来看此时存在的情况:

  • 如果 isOnSyncQueue 返回 false ,即 线程不在AQS的队列中 ,进入自旋,调用 LockSupport#park 暂停线程;
  • 如果 isOnSyncQueue 返回 true ,即 线程在AQS的队列中 ,不进入自旋,执行后续逻辑。

结合注释1和注释2的部分, Condition#await 的实现原理了就很清晰了:

  • Condition 与AQS分别维护了一个等待队列,而且是互斥的,即 同一个节点只会出现在一个队列中
  • 当调用 Condition#await 时,将线程添加到 Condition 的队列中(注释1),同时从AQS队列中移出(注释2);
  • 接着判断线程位于的队列:
    • 位于 Condition 队列中,该线程需要被暂停,调用 LockSupport#park
    • 位于AQS队列中,该线程正在等待获取锁。

基于以上的结论,我们已经能够猜到唤醒方法 Condition#signalAll 的原理了:

  • 将线程从 Condition 队列中移出,并添加到AQS的队列中;
  • 调用 LockSupport.unpark 唤醒线程。

至于这个猜想是否正确,我们接着来看唤醒方法的实现.

Tips :如果忘记了AQS中相关方法是如何实现的,可以回顾下《 AQS的今生,构建出JUC的基础 》.

signal和signalAll方法的实现

来看 signal 和 signalAll 的源码:

                        
                          // 唤醒一个处于等待中的线程
public final void signal() {
  if (!isHeldExclusively()) {
    throw new IllegalMonitorStateException();
  }
  // 获取Condition队列中的第一个节点
  Node first = firstWaiter;
  if (first != null) {
    // 唤醒第一个节点
    doSignal(first);
  }
}

// 唤醒全部处于等待中的线程
public final void signalAll() {
    if (!isHeldExclusively()){
      throw new IllegalMonitorStateException();
    }
        
    Node first = firstWaiter;
    if (first != null) {
      // 唤醒所有节点
      doSignalAll(first);
    }  
}

                        
                      

两个方法唯一的差别在于头节点不为空的场景下,是调用 doSignal 唤醒一个线程还是调用 doSignalAll 唤醒所有线程:

                        
                          private void doSignal(Node first) {
  do {
    // 更新头节点
    if ( (firstWaiter = first.nextWaiter) == null) {
      // 无后继节点的场景
      lastWaiter = null;
    }
    // 断开节点的连接
    first.nextWaiter = null;
    // 唤醒头节点
  } while (!transferForSignal(first) && (first = firstWaiter) != null);
}

private void doSignalAll(Node first) {
  // 将Condition的队列置为空
  lastWaiter = firstWaiter = null;
  do {
    // 断开链接
    Node next = first.nextWaiter;
    first.nextWaiter = null;
    // 唤醒当前头节点
    transferForSignal(first);
    // 更新头节点
    first = next;
  } while (first != null);
}

                        
                      

可以看到,无论是 doSignal 还是 doSignalAll 都只是将节点移出 Condition 队列,而真正起到唤醒作用的是 transferForSignal 方法,从方法名可以看到该方法是通过“ 转移 ”进行唤醒的,我们来看源码:

                        
                          final boolean transferForSignal(Node node) {
  // 通过CAS替换node的状态
  // 如果替换失败,说明node不处于Node.CONDITION状态,不需要唤醒
  if (!node.compareAndSetWaitStatus(Node.CONDITION, 0)) {
    return false;
  }
  // 将节点添加到AQS的队列的队尾
  // 并返回老队尾节点,即node的前驱节点
  Node p = enq(node);
  int ws = p.waitStatus;
  // 对前驱节点状态的判断
  if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL)) {
    LockSupport.unpark(node.thread);
  }
  return true;
}

                        
                      

transferForSignal 方法中,调用 enq 方法将 node 重新添加到AQS的队列中,并返回 node 的前驱节点,随后对前驱节点的状态进行判断:

  • \(ws > 0\) 时,前驱节点处于 Node.CANCELLED 状态,前驱节点退出锁的争抢, node 可以直接被唤醒;
  • \(ws \leq 0\) 时,通过CAS修改前驱节点的状态为 Node.SIGNAL ,设置失败时,直接唤醒 node

《 AQS的今生,构建出JUC的基础 》中介绍了 waitStatus 的5种状态,其中 Node.SIGNAL 状态表示需要唤醒后继节点。另外,在分析 shouldParkAfterFailedAcquire 方法的源码时,我们知道在进入AQS的等待队列时,需要将前驱节点的状态更新为 Node.SIGNAL .

最后来看 enq 的实现:

                        
                          private Node enq(Node node) {
  for (;;) {
    // 获取尾节点
    Node oldTail = tail;
    if (oldTail != null) {
      // 更新当前节点的前驱节点
      node.setPrevRelaxed(oldTail);
      // 更新尾节点
      if (compareAndSetTail(oldTail, node)) {
        oldTail.next = node;
        // 返回当前节点的前驱节点(即老尾节点)
        return oldTail;
      }
    } else {
      initializeSyncQueue();
    }
  }
}

                        
                      

enq 的实现就非常简单了,通过CAS更新AQS的队列尾节点,相当于添加到AQS的队列中,并返回尾节点的前驱节点。好了,唤醒方法的源码到这里就结束了,是不是和我们当初的猜想一模一样呢?

图解ConditionObject原理

功能上, Condition 实现了AQS版 Object#wait 和 Object#notify ,用法上也与之相似,需要先获取锁,即需要在 lock 与 unlock 之间调用。原理上, 简单来说就是线程在AQS的队列和 Condition 的队列之间的转移 .

线程t持有锁

假设有线程t已经获取了 ReentrantLock ,线程t1,t2和t3正在AQS的队列中等待,我们可以得到这样的结构:

线程t执行Condition#await

如果线程t中调用了 Condition#await 方法,线程t进入 Condition 的等待队列中,线程t1获取 ReentrantLock ,并从AQS的队列中移出,结构如下:

线程t1执行Condition#await

如果线程t1中也执行了 Condition#await 方法,同样线程t1进入 Condition 队列中,线程t2获取到 ReentrantLock ,结构如下:

线程t2执行Condition#signal

如果线程t2执行了 Condition#signal ,唤醒 Condition 队列中的第一个线程,此时结构如下:

通过上面的流程,我们就可以得到线程是如何在 Condition 队列与AQS队列中转移的:

结语

关于 Condition 的内容到这里就结束了,无论是理解,使用还是剖析原理, Condition 的难度并不高,只不过大家可能平时用得比较少,因此多少有些陌生.

最后,截止到文章发布,我应该是把开头两道题目的 题解 写完了吧~~ 。


好了,今天就到这里了,Bye~~ 。

最后此篇关于17.AQS中的Condition是什么?的文章就讲到这里了,如果你想了解更多关于17.AQS中的Condition是什么?的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。

30 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com