Java中的线程协作之Condition-低调大师

Java中的线程协作之Condition

2019-04-16 597

Java中的线程协作之Condition
一、Condition接口
1、Condition接口的常用方法介绍
复制代码
1 /**
2 * 已经获取到锁的线程调用该方法会进入等待状态，知道其他持有锁的线程通知（signal）等待队列中的线程或者被中断退出等待队列；
3 * 如果该线程已经从该方法中返回，表名线程已经获取到了Condition对象对应的锁
4 */
5 public final void await() throws InterruptedException {...}
6 /**
7 * 还是进入等待状态的方法，只是该方法对中断不敏感：当前调用该方法的线程只有被通知（signal）才能从等待队列中退出
8 */
9 public final void awaitUninterruptibly() {...}
10 /**
11 * 当前线程进入等待状态，被通知、中断或者超时之后被唤醒。返回值就是表示剩余的时间，即
12 * 如果在nanosTimeout纳秒之前被唤醒，返回值就是实际耗时；如果返回值是0或者负数，就认为是超时了
13 */
14 public final long awaitNanos(long nanosTimeout) {...}
15 /**
16 * 调用该方法的线程会进入等待状态直到被通知、中断或者到达某个超时时间。
17 * 意味着没有到达指定的某个时间被通知，就会返回true;如果到达指定时间，返回false
18 */
19 public final boolean awaitUntil(Date deadline){}
20 /**
21 * 当前持有Condition对象对应锁的线程，调用该方法之后会唤醒一个等待在Condition上的线程
22 */
23 public final void signal() {}
24 /**
25 * 当前持有Condition对象对应锁的线程，调用该方法之后会唤醒等待在Condition上的所有线程
26 */
27 public final void signalAll() {}
复制代码
　　Condition的使用模板：Condition的获取必须通过Lock的newCondition方法，表示Condition对象与该锁关联，一般讲Condition对象作为成员变量，调用上面的await方法之后当前线程才会释放锁并在等待队列中进行等待；当其他的线程（在没有中断的情况下）调用该condition对象的signal方法的时候就会通知等待队列中的等待线程从await方法返回（返回之前已经获取锁）。

复制代码
1 Lock lock = new ReentrantLock();
2 Condition con = lock.newCondition();
3 public void conWait() {
4 lock.lock();
5 try {
6 con.await();
7 } catch(InterruptedException e) {
8 ...
9 }finally {
10 lock.unlock();
11 }
12 }
13
14 public void conSignal() {
15 lock.lock();
16 try {
17 con.signal();
18 } catch(InterruptedException e) {
19 ...
20 }finally {
21 lock.unlock();
22 }
23 }
复制代码
2、Condition的实现分析
a）源码流程分析
　　我们通过跟踪源码可以看出来，首先创建锁对象（new ReentrantLock()），然后根据锁对象关联响应的Condition对象，然后通过Condition对象中维护的等待队列实现等待(await)通知(signal)机制。

复制代码
1 public Condition newCondition() { //ReentrantLock类中的方法
2 return sync.newCondition();
3 }
4 //ConditionObject类实现Condition接口，除此室外ConditionObject也是AQS的一个内部类，Condition的操作需要与锁关联起来
5 final ConditionObject newCondition() {
6 return new ConditionObject();
7 }
8 //AQS的内部类ConditionObject，其中维护了一个等待队列，通过该队列实现等待通知机制
9 public class ConditionObject{
10 /**
11 * 返回等待队列中的线程集合
12 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
13 * returns {@code false}
14 */
15 protected final Collection getWaitingThreads() {
16 if (!isHeldExclusively())
17 throw new IllegalMonitorStateException();
18 ArrayList list = new ArrayList();
19 for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
20 if (w.waitStatus == Node.CONDITION) {
21 Thread t = w.thread;
22 if (t != null)
23 list.add(t);
24 }
25 }
26 return list;
27 }
28 }
复制代码
b）具体实现
　　上面说到了Condition是通过等待队列来实现等待通知功能的，那么就分析等待队列和等待通知机制的实现

①等待队列实现
　　等待队列是一个FIFO的队列，其中每个结点都包含一个处于Condition对象上等待的线程引用（当一个获取到锁的线程调用await方法，就会释放锁资源，被包装成一个Node然后添加到等待队列中进入等待状态；这里面的Node结点还是和AQS中的实现机理一样，Node是AQS中的静态内部类）。

　　ConditionObject类中有下面两个属性，分别代表一个Condition对应的等待队列的首节点和尾结点。当前线程调用await方法之后就会被构造成一个Node结点然后加入到等待队列的尾部。

1 /* Condition等待队列头结点 /
2 private transient Node firstWaiter;
3 /* Condition等待队列尾结点 /
4 private transient Node lastWaiter;
　　下面是等待队列的基本结构，Condition对象中有首尾结点的引用。新增加的结点需要将原有的尾结点的下一节点指向它，然后更新lastWaiter即可。

　　上面的情况是一个Condition对象对应一个等待队列和一个同步队列（上面新添加的Node3就是从同步队列中移除然后添加过来的），在同步器组件实现中，会拥有一个同步队列和多个等待队列。

②等待操作的实现
　　持有锁的线程调Condition的await方法之后会释放锁，然后进入等待状态。既然是持有锁的线程，那么该线程应该位于同步队列的首节点位置，其调用await方法之后就会从同步队列首节点移到等待队列的尾结点等待。具体将其移到等待队列是addConditionWaiter方法实现。下面是await方法和addConditionWaiter方法的实现分析。

复制代码
public final void await() throws InterruptedException {

if (Thread.interrupted())
    throw new InterruptedException();
Node node = addConditionWaiter(); //将当前线程加入等待队列
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);

}
private Node addConditionWaiter() {

Node t = lastWaiter;
/**
 * waitStatus值表示线程正在等待条件（原本结点在等待队列中，结点线程等待在Condition上，当其他线程对
 * Condition调用了signal()方法之后）该结点会从等待队列中转移到同步队列中，进行同步状态的获取 
 * static final int CONDITION = -2;
*/
if (t != null && t.waitStatus != Node.CONDITION) {
    unlinkCancelledWaiters();
    t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION); //构造成Condition的等待队列中的对应的结点
//增加的结点需要将原有的尾结点的下一节点指向它，然后更新lastWaiter
if (t == null)
    firstWaiter = node;
else
    t.nextWaiter = node;
lastWaiter = node;
return node;

}
复制代码

③通知操作的实现
　　通知操作的实现机制就是将当前等待队列中的首节点中的线程唤醒，将其加入同步队列中。

复制代码
1 public final void signal() {
2 if (!isHeldExclusively()) //检查当前线程是否获取锁
3 throw new IllegalMonitorStateException();
4 Node first = firstWaiter;
5 if (first != null)
6 doSignal(first);
7 }
复制代码
　　唤醒线程使其进入同步队列之后，我们再来看await方法中那些没有执行的代码。

复制代码
1 public final void await() throws InterruptedException {
2 if (Thread.interrupted())
3 throw new InterruptedException();
4 Node node = addConditionWaiter(); //将当前线程加入等待队列
5 int savedState = fullyRelease(node); //释放当前线程持有的锁
6 int interruptMode = 0;
7 //根据下面的源码可以看出，当前线程如果掉用await方法之后会进入等待队列，那么在退出等待队列之前会一直执行这个循环
8 while (!isOnSyncQueue(node)) {
9 LockSupport.park(this); //唤醒节点中的线程
10 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
11 break;
12 }
13 //通过acquireQueued源码可以发现，获取锁的流程和ReentrantLock这种独占式获取同步状态的流程基本一致
14 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
15 interruptMode = REINTERRUPT;
16 if (node.nextWaiter != null) // clean up if cancelled
17 unlinkCancelledWaiters();
18 if (interruptMode != 0)
19 reportInterruptAfterWait(interruptMode);
20 }
21 final boolean isOnSyncQueue(Node node) {
22 if (node.waitStatus == Node.CONDITION || node.prev == null) //判断当前队列是否在等待队列中
23 return false;
24 if (node.next != null) // If has successor, it must be on queue
25 return true;
26 return findNodeFromTail(node);
27 }
28 //竞争锁资源的同步队列
29 final boolean acquireQueued(final Node node, int arg) {
30 boolean failed = true;
31 try {
32 boolean interrupted = false;
33 for (;;) {
34 final Node p = node.predecessor(); //得到当前结点的前驱结点
35 if (p == head && tryAcquire(arg)) { //当前结点的前驱结点为头结点，并且尝试获取锁成功
36 setHead(node); //将当前获取到锁的结点设置为头结点
37 p.next = null; // help GC
38 failed = false;
39 return interrupted;
40 }
41 //如果获取同步状态失败，应该自旋等待继续获取并且校验自己的中断标志位信息
42 if (shouldParkAfterFailedAcquire(p, node) &&
43 parkAndCheckInterrupt())
44 interrupted = true;
45 }
46 } finally {
47 if (failed)
48 cancelAcquire(node);
49 }
50 }
复制代码
　　从上面的代码中我们可以看出，当调用await方法的线程在没有回到同步队列之前，都会一直在while (!isOnSyncQueue(node)){...}循环中，只有被唤醒退出等待队列进入同步队列才会从循环中退出；之后调用acquireQueued()开始自旋等待锁的获取，这个自旋的过程和前面介绍的AQS中独占式锁的获取流程一样；最后，如果线程从这个自旋的过程退出了，就代表当前线程再次获取了锁资源，最后也从await方法返回。所以，一个线程调用await方法之后，只有最终获取到锁才会从该方法返回。而对于signalAll而言就是对等待队列中的每个线程通知（signal）一次，这样就可以将等待队列中的所有线程移到同步队列中进行锁资源的获取。

回到顶部
二、Condition接口使用
1、Condition接口搭配ReentrantLock实现生产者消费者模式
复制代码
1 package cn.source.condition;
2
3 import java.util.LinkedList;
4 import java.util.concurrent.TimeUnit;
5 import java.util.concurrent.locks.Condition;
6 import java.util.concurrent.locks.Lock;
7 import java.util.concurrent.locks.ReentrantLock;
8
9 public class ConditionProducerAndConsumer {
10
11 private LinkedList list = new LinkedList();
12 private static final int MAX_NUM = 10; //容器的最大数量
13 private int count = 0; //容器中实际数量
14
15 private Lock lock = new ReentrantLock();
16 private Condition producer = lock.newCondition();
17 private Condition consumer = lock.newCondition();
18
19 private int getCount() {
20 return count;
21 }
22
23 private void put(E e) {
24 lock.lock(); //首先需要获取锁
25 try {
26 //这里是判断容器是否已满，注意需要使用while:如果使用if的话可能导致所有的消费线程都处于等待状态
27 while(list.size() == MAX_NUM) {
28 System.out.println(Thread.currentThread().getName() + "正在等待中");
29 producer.await(); //生产者线程进入等待状态
30 }
31 //添加元素
32 list.add(e);
33 count ++;
34 consumer.signalAll();//将消费者线程唤醒
35 } catch (InterruptedException e1) {
36 e1.printStackTrace();
37 } finally {
38 lock.unlock();
39 }
40 }
41
42 private E get() {
43 E e = null;
44 lock.lock();
45 try {
46 while(list.size() == 0) {
47 System.out.println(Thread.currentThread().getName() + "正在等待");
48 consumer.await(); //消费者线程进入等待状态
49 }
50 e = list.removeFirst();
51 count --;
52 producer.signalAll(); //消费元素之后，将生产者线程唤醒
53 } catch (InterruptedException e1) {
54 e1.printStackTrace();
55 } finally {
56 lock.unlock();
57 }
58 return e;
59 }
60
61 public static void main(String[] args) {
62 SyncProducerAndConsumer syncProducerAndConsumer = new SyncProducerAndConsumer<>();
63 for (int i = 0; i < 10; i++) { //开启10个线程
64 new Thread(new Runnable() {
65 @Override
66 public void run() {
67 for (int j = 0; j < 5; j++) { //每个线程从容器中获取5次数据
68 System.out.println(syncProducerAndConsumer.get());
69 }
70 }
71
72 }, "消费者线程" + i).start();;
73 }
74 //休眠2秒，所有的消费者线程都已经启动并且处于等待状态
75 try {
76 TimeUnit.SECONDS.sleep(2);
77 } catch (InterruptedException e) {
78 e.printStackTrace();
79 }
80
81 for (int i = 0; i < 2; i++) { //开启两个生产者线程
82 new Thread(new Runnable() {
83 @Override
84 public void run() {
85 for (int j = 0; j < 25; j++) { //每个生产者线程想容器中添加25个数据，当容器中数据到达10个的时候生产者线程会阻塞
86 syncProducerAndConsumer.put("add value " + j);
87 }
88 }
89 }, "生产者线程"+i).start();
90 }
91 }
92
93 }
复制代码
2、synchronized组合wait/notify实现生产者消费者模式
复制代码
1 package cn.source.condition;
2
3 import java.util.LinkedList;
4 import java.util.concurrent.TimeUnit;
5
6 public class SyncProducerAndConsumer {
7
8 private LinkedList list = new LinkedList();
9 private static final int MAX_NUM = 10; //容器的最大数量
10 private int count = 0; //容器中实际数量
11
12 public synchronized int getCount() {
13 return count;
14 }
15
16 public synchronized void put(E e) {
17 while(list.size() == MAX_NUM) { //这里是判断容器是否已满，注意需要使用while:如果使用if的话可能导致所有的消费线程都处于等待状态
18 try {
19 this.wait(); //容器满了之后，生产者线程进入等待状态
20 } catch (InterruptedException e1) {
21 e1.printStackTrace();
22 }
23 }
24 //容器未满，生产者线程就想容器中添加数据
25 list.add(e);
26 count ++;
27 this.notifyAll(); //此时容器中已经存在数据，唤醒等待的消费者线程
28 }
29
30 public synchronized E get() {
31 E e = null;
32 while(list.size() == 0) { //判断容器是否为空，如果为空就进入等待状态，这里也使用while
33 try {
34 this.wait();
35 } catch (InterruptedException e1) {
36 e1.printStackTrace();
37 }
38 }
39 e = list.removeFirst();
40 count --;
41 this.notifyAll();
42 return e;
43 }
44
45 public static void main(String[] args) {
46 SyncProducerAndConsumer syncProducerAndConsumer = new SyncProducerAndConsumer<>();
47 for (int i = 0; i < 10; i++) { //开启10个线程
48 new Thread(new Runnable() {
49 @Override
50 public void run() {
51 for (int j = 0; j < 5; j++) { //每个线程从容器中获取5次数据
52 System.out.println(syncProducerAndConsumer.get());
53 }
54 }
55
56 }, "消费者线程" + i).start();;
57 }
58 //休眠2秒，所有的消费者线程都已经启动并且处于等待状态
59 try {
60 TimeUnit.SECONDS.sleep(2);
61 } catch (InterruptedException e) {
62 e.printStackTrace();
63 }
64
65 for (int i = 0; i < 2; i++) { //开启两个生产者线程
66 new Thread(new Runnable() {
67 @Override
68 public void run() {
69 for (int j = 0; j < 25; j++) { //每个生产者线程想容器中添加25个数据，当容器中数据到达10个的时候生产者线程会阻塞
70 syncProducerAndConsumer.put("add value " + j);
71 }
72 }
73 }, "生产者线程"+i).start();
74 }
75 }
76
77
78 }
复制代码
3、Object中的等待唤醒机制和Condition的等待通知机制对比
原文地址https://www.cnblogs.com/fsmly/p/10721459.html

微信关注我们

原文链接：https://yq.aliyun.com/articles/698605

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

C++之父：精通C++很难，但你一天之内就能学习使用C++

精通C++听起来好像就是一个笑话。为什么C++比别的语言难学那么多？其实这基本上是因为C++之父Bjarne Stroustrup 说过的一句话“我特别的讨厌语言的设计者把自己的喜好强加给用户”（看向go）。结果C++为了不限制你的想法，于是也就变成了现在的样子-- ，那么如何学好C++，甚至是出色掌握C++? 1.谈谈C++的书籍。没有，也不可能有一本书对于所有人来说都是最好的。不过对于那些真正的程序员来说，如果他喜欢从“经典风格”的书中间学习一些新的概念和技术，我推荐我的The C++ Programming Language, 1998年的第三版和特别版。那本书讲的是纯而又纯的C++，完全独立于平台和库（当然得讲到标准库）。该书面向那些有一定经验的程序员，帮助他们掌握C++，但不适合毫无经验的初学者入门，也不适合那些临时程序员品尝C++快餐。所以这本书的重点在于概念和技术，而且在完整性和精确性上下了不少功夫。学习C++要花多长时间？这要看你说的“学习”是什么意思了。如果你是一个Pascal程序员，你应该能很快地使你的C++水平达到与Pascal相近的程度；而如果你是一个C程...

2019-04-17

756

好程序员分享大势所趋 HTML5成Web开发者最关心的技术，最近，在Stack Exchange上出现了一个比较热门的问题：Web开发者最头疼的问题是什么?结果并不是大家通常认为的兼容性问题，而是关于HTML5。在所有与前端开发相关的技术中，如JavaScript、CSS3、HTML5等，Web开发者们私下最经常讨论的技术领域是哪一个呢?理解这个问题就可以更好地了解当前在Web技术领域，大家所面临的挑战是什么?为了一探究竟，英属哥伦比亚大学就“Web开发者经常挖掘的问题”展开了研究。他们在Stacke Exchange上检查了50多万个与JavaScript、HTML5、CSS等与Web有关的技术问题，对这些问题的答案进行了分析，他们总结出了哪些问题是开发者们最感兴趣和经常询问的，总结出了Web开发者们目前所面临的挑战。 JavaScript仍然称王 CSS和HTML5在迅速崛起纵观最流行的三大Web技术(JavaScript、HTML5、CSS)，JavaScript仍然是开发者们最关心的话题。但在Stack Exchange上，HTML5、CSS有关的问题数量也在迅速上升。研...

2019-04-17

563

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。