0
点赞
收藏
分享

微信扫一扫

并发工具类：Condition如何实现条件通知？

MaxWen 2022-08-07 阅读 57
java开发语言后端等待队列通知机制编程语言
加载失败


并发工具类：Condition如何实现条件通知？_通知机制

AQS如何实现等待通知？

在进行线程间的通信时，可以用基于Object对象的wait和notify方法实现等待/通知机制

基于Object实现等待/通知机制的相关方法

方法名称

描述

notify()

通知一个在对象上等待的线程，使其从wait()方法返回

notifyAll()

通知所有等待在该对象上的线程

wait()

调用该方法的线程进入WAITING状态，只有等待另外线程的通知或被中断才会返回，需要注意，调用wait()方法后，会释放对象的锁

wait(long)

超时等待一段时间，这里的参数是毫秒，也就是等待长达n毫秒，如果没有通知就超时返回

wait(long, int)

对于超时时间更细粒度的控制，可以达到纳秒

用例子演示一下用法

public class WaitNotify {

    private static Object lock = new Object();

    public static void main(String[] args) throws InterruptedException {
        Thread waitThread = new Thread(new Wait());
        waitThread.start();
        TimeUnit.SECONDS.sleep(1);
        Thread notifyThread = new Thread(new Notify());
        notifyThread.start();
    }

    private static class Wait implements Runnable {

        @Override
        public void run() {
            synchronized (lock) {
                try {
                    System.out.println("lock wait start");
                    lock.wait();
                    System.out.println("lock wait end");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 条件满足，完成工作
                System.out.println("run finish");
            }
        }
    }

    private static class Notify implements Runnable {

        @Override
        public void run() {
            synchronized (lock) {
                System.out.println("lock notify start");
                lock.notify();
                System.out.println("lock notify end");
            }
        }
    }
}

lock wait start
lock notify start
lock notify end
lock wait end
run finish

wait()方法被执行后，锁被自动释放，但notify()方法被执行后，锁却不自动释放，必须执行完notify()方法所在的同步synchronized代码块后才释放锁

在AQS相关类中怎么实现这种等待/通知机制呢？

答案是Condition，Condition是一个接口，AbstractQueuedSynchronizer中有一个内部类实现了这个接口，一个Condition就代表一个条件队列

基于Condition实现等待/通知机制

和synchronized一样，调用await和signal方法时，必须获得与Condition相关的锁

Condition的主要方法有2个

  1. await，将线程放入等待队列进行等待，当其他线程调用signal或者interrupt方法时会被唤醒
  2. signal，唤醒等待队列中的线程，将其放入同步队列，获得锁后开始执行

Conditon使用例子如下

public class WaitNotifyUseCondition {

    private static ReentrantLock lock = new ReentrantLock();
    private static Condition conditionA  = lock.newCondition();
    private static Condition conditionB = lock.newCondition();

    public static void main(String[] args) throws InterruptedException {
        Thread waitThreadA = new Thread(new WaitA());
        waitThreadA.start();
        Thread waitThreadB = new Thread(new WaitB());
        waitThreadB.start();
        TimeUnit.SECONDS.sleep(2);
        lock.lock();
        try {
            conditionA.signal();
        } finally {
            lock.unlock();
        }
    }

    private static class WaitA implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println("WaitA begin wait");
                conditionA.await();
                System.out.println("WaitA end wait");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    private static class WaitB implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println("WaitB begin wait");
                conditionB.await();
                System.out.println("WaitB end wait");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }
}

WaitA begin wait
WaitB begin wait
WaitA end wait

WaitThreadB因为没有被通知，一直阻塞

可以看到Condition比Object的等待通知更强大，因为它可以实现条件通知

其实原因也和简单，基于Object的等待通知只有一个等待队列，而在AQS中有多个等待队列，一个Condition类代表一个条件队列

同步队列和等待队列

并发工具类：Condition如何实现条件通知？_后端_02


在之前的AQS文章中，我们提到了同步队列，本节我们又提到了等待队列，那他们两者是如何协同工作的？

AbstractQueuedSynchronizer内部维护着一个同步队列（双向链表实现），多个条件队列（单向链表实现），条件队列由AbstractQueuedSynchronizer的内部类ConditionObject来维护，new一个ConditonObject ，则多一个条件队列，当一个线程执行await方法是，会把当线程包装成一个Node节点，放到执行await方法的ConditionObject的条件队列中，释放锁并被阻塞，当执行signal方式时，会把条件队列的第一个节点移除，并转移到同步队列中，获取到锁即可继续执行

ConditionObject 是AbstractQueuedSynchronizer的一个内部类，用来实现条件队列，属性如下

public class ConditionObject implements Condition, java.io.Serializable {

  // 条件队列的头节点
  private transient Node firstWaiter;
  
  // 条件队列的尾节点
  private transient Node lastWaiter;

  public ConditionObject() { }

  // 阻塞过程中不响应中断，仅设置标志位，让之后的方法处理
  private static final int REINTERRUPT =  1;
  
  // 阻塞过程中响应中断，并throw InterruptedException
  private static final int THROW_IE    = -1;

}

假如在阻塞过程中发生了中断，REINTERRUPT标志了中断发生在 signalled之后，
THROW_IE标志了中断发生在 signalled之前，从而决定采用那种方式响应中断

await阻塞线程

将线程放入等待队列，并且阻塞线程。当调用signal的时候，唤醒线程，进入同步队列获取锁

// ConditionObject
public final void await() throws InterruptedException {
  if (Thread.interrupted())
    throw new InterruptedException();
  // 当前线程加入等待队列
  Node node = addConditionWaiter();
  // 挂起线程前，必须释放当前锁，这里调用ReentrantLock释放锁的逻辑
  int savedState = fullyRelease(node);
  // 标志位
  int interruptMode = 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);
}

增加等待节点的逻辑，每次增加节点都会剔除非condition的节点

// ConditionObject
private Node addConditionWaiter() {
  Node t = lastWaiter;
  // If lastWaiter is cancelled, clean out.
  if (t != null && t.waitStatus != Node.CONDITION) {
    // 清除等待队列中取消状态的节点
    unlinkCancelledWaiters();
    t = lastWaiter;
  }
  Node node = new Node(Thread.currentThread(), Node.CONDITION);
  // 链表还没有初始化
  if (t == null)
    firstWaiter = node;
  else
    t.nextWaiter = node;
  lastWaiter = node;
  return node;
}

释放锁

// AbstractQueuedSynchronizer
final int fullyRelease(Node node) {
  boolean failed = true;
  try {
    int savedState = getState();
    if (release(savedState)) {
      failed = false;
      return savedState;
    } else {
      // 释放锁失败
      throw new IllegalMonitorStateException();
    }
  } finally {
    // 释放锁失败后，将当前节点状态设置为CANCELLED
    // 后序会被清理出条件队列
    if (failed)
      node.waitStatus = Node.CANCELLED;
  }
}

判断节点是否在同步队列

// AbstractQueuedSynchronizer
final boolean isOnSyncQueue(Node node) {
  // 节点在条件队列
  // 同步队列中节点的状态 只能为0、SIGNAL、PROPAGATE 和 CANCELLED 其中之一
  if (node.waitStatus == Node.CONDITION || node.prev == null)
    return false;
  // 如果后继节点不为null,则表明节点在同步队列上
  // 因为条件队列使用的是nextWaiter指向后继节点的
  // 条件队列上节点的next均为null
  if (node.next != null) // If has successor, it must be on queue
    return true;
  
  // 走到这一步，说明node.prev!=null && node.next=null
  // 但这并不能说明node在同步队列中，因为节点在入队过程中
  // 是先设置node.prev后设置node.next的（详见addWaiter方法）
  // 有可能CAS设置尾节点失败，导致没有加入队列
  // 所以从尾到头遍历一遍
  return findNodeFromTail(node);
}

// AbstractQueuedSynchronizer
private boolean findNodeFromTail(Node node) {
  Node t = tail;
  for (;;) {
    if (t == node)
      return true;
    if (t == null)
      return false;
    t = t.prev;
  }
}

检测线程在等待期间是否发生中断

// ConditionObject
// Checks for interrupt, returning THROW_IE if interrupted
// before signalled, REINTERRUPT if after signalled, or
// 0 if not interrupted.
private int checkInterruptWhileWaiting(Node node) {
  return Thread.interrupted() ?
    (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
    0;
}

// AbstractQueuedSynchronizer
final boolean transferAfterCancelledWait(Node node) {
  // signalled之前发生中断,因为signalled之后会将会将节点状态从CONDITION 设置为0
  if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    enq(node);
    return true;
  }

  // signalled之后发生中断
  // 如果节点还没有被放入同步队列，则放弃当前CPU资源
  // 让其他任务执行
  while (!isOnSyncQueue(node))
    Thread.yield();
  return false;
}

清除等待队列中取消状态的节点

// ConditionObject
private void unlinkCancelledWaiters() {
  Node t = firstWaiter;
  // 指向尾节点
  Node trail = null;
  while (t != null) {
    Node next = t.nextWaiter;
    if (t.waitStatus != Node.CONDITION) {
      t.nextWaiter = null;
      // 只有头节点的状态不是CONDITION才会执行到这一步
      if (trail == null)
        firstWaiter = next;
      else
        trail.nextWaiter = next;
      // 遍历完链表，设置尾节点
      if (next == null)
        lastWaiter = trail;
    }
    else
      trail = t;
    t = next;
  }
}

响应中断的方式
interruptMode=THROW_IE（-1）：await退出时，直接抛出InterruptedException
interruptMode=REINTERRUPT（1）：await退出时，重置中断标记位

// ConditionObject
private void reportInterruptAfterWait(int interruptMode)
  throws InterruptedException {
  if (interruptMode == THROW_IE)
    // 直接响应中断
    throw new InterruptedException();
  else if (interruptMode == REINTERRUPT)
    // 重置中断标记位
    selfInterrupt();
}

// AbstractQueuedSynchronizer
static void selfInterrupt() {
  Thread.currentThread().interrupt();
}

signal，唤醒等待时间最长的线程

// ConditionObject
public final void signal() {
  // 当前线程没有获取到锁
  if (!isHeldExclusively())
    throw new IllegalMonitorStateException();
  // 唤醒等待队列中的头结点
  Node first = firstWaiter;
  if (first != null)
    doSignal(first);
}

// ConditionObject
private void doSignal(Node first) {
  do {
    // 将同步队列的头结点，设置为目前头结点的下一个节点
    // 如果头节点的下一个节点为null，则设置尾节点为null
    if ( (firstWaiter = first.nextWaiter) == null)
      lastWaiter = null;
    // 将first节点从条件队列中移除
    first.nextWaiter = null;
    // 通知第一个非CANCELLED节点被唤醒，或者遍历完，退出
  } while (!transferForSignal(first) &&
       (first = firstWaiter) != null);
}

唤醒第一个非CANCELLED节点

// 将节点从条件队列放入同步队列，true为成功 
// AbstractQueuedSynchronizer
final boolean transferForSignal(Node node) {
   
  // 通过CAS将节点的状态从CONDITION设置为0
  // 如果设置失败，说明这个节点状态为CANCELLED
  if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
    return false;

  // 加入同步队列，并返回前继节点
  Node p = enq(node);
  int ws = p.waitStatus;
  // 前继节点为CANCELLED状态，或者设置SIGNAL状态失败
  if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
    // 唤醒线程
    LockSupport.unpark(node.thread);
  return true;
}

signalAll和signal实现类似，区别如下，signal将等待队列中的一个非CANCELLED节点放到同步队列，而signalAll是将等待队列中的所有非CANCELLED节点放到同步队列中

参考博客

队列示意图
[1]http://www.importnew.com/9281.html
[2]https://www.jianshu.com/p/6b5aa7b7684c
[3]http://www.importnew.com/9281.html


举报

相关推荐

并发编程--并发编程包Condition条件

Condition阻塞队列并发编程阻塞状态Linux系统/运维

JUC（2）使用condition实现精准通知唤醒

javai++Linux系统/运维

并发编程-Condition

J2SE-并发编程RxJavaJava服务器端编程编程语言爱好者Java 核心技术Java

并发编程 -- condition

通知机制等待队列等待状态Windows系统/运维

【并发编程】并发工具类

java后端共享锁Systemi++系统/运维

java并发之ReentrantLock.Condition实现原理

java开发语言后端

Java并发之Condition

队列多线程java并发编程threadPython后端开发

TP5 condition 多个条件如何写

前端页面公众号javascript前端开发

条件变量(condition_variable)

c++

条件变量-condition_variable

多线程条件变量#include子线程Html/CSS前端开发
0 条评论
加载失败