一、协调生产/消费的需求

本文内容主要想向大家介绍一下Lock结合Condition的使用方法,为了更好的理解Lock锁与Condition锁信号,我们来手写一个ArrayBlockingQueue。 JDK实际上已经有这个类,基于Lock锁与Condition锁信号实现的,当然JDK实现代码很复杂包含了更严谨的逻辑校验,以及从性能优化的角度做了更多的工作。本文中我们只是来简单实现一下其核心逻辑:

  • ArrayBlockingQueue初始化构造时指定容量上限最大值
  • 提供put方法,当达到Queue队列容量上限最大值,阻塞生产数据的线程。
  • put方法生产数据之后,队列肯定是不为空,通知消费者线程进行消费。
  • 提供take方法,当Queue队列容量为0时候,阻塞消费数据的线程。
  • take方法执行之后,队列肯定不是满的,通知生产者线程进行生产。
  • 一条数据只能被take一次,take之后数据从queue中删除

相信实现完成上面的逻辑之后,java并发编程之Lock锁与Condition锁信号,你肯定是掌握了!其实这个逻辑基本上就是kafka生产者客户端缓冲队列,批量进行数据发送的实现逻辑。区别是take方法一次取出缓冲区所有数据,本文take方法一次取出一条数据。

二、构造方法

构造队列的方法很简单,使用一个List作为数据存储队列,并指定其容量。到此我们还没有实现容量判断,以及阻塞线程的功能。

//类成员变量-存储数据的队列
private List<Object> queue;
//类成员变量-存储队列的容量上限
private int queueSize; public MyBlockingQueue(int queueSize) {
this.queueSize = queueSize;
queue = new ArrayList<>(queueSize);//存储消息的集合
}

三、Lock& Condition逻辑设计

首先我们要有一把锁,保证数据put与take操作的同步性,即:一条数据只能被take一次,take之后数据从queue中删除;以及创建Condition逻辑都需要Lock锁。学过java基础并发编程的同学,可以把Lock锁理解为Synchronized 同步代码块功能是一样的。我写过一个专栏《java并发编程》中介绍了二者的区别,欢迎关注。

private Lock lock = new ReentrantLock();//锁

Condition逻辑大家可以理解为传统JDK多线程编程中的wait与notify,但是Condition的语义更容易被理解。如下文代码所示:

private Condition notFull = lock.newCondition();   //队列不为满

notFull.signal();   //通知生产者队列不为满,可以继续生产数据(通常在消费者拿走数据之后,调用)
notFull.await(); //队列已满,阻塞生产线程(await对condition逻辑取反)
private Condition notEmpty = lock.newCondition();  //队列不为空

notEmpty.signal();  //通知消费者线程队列不为空,可以继续消费数据(通常在生产者生产数据之后,调用)
notEmpty.await(); //队列已经空了,阻塞消费线程(await对condition逻辑取反)

大家在使用Lock& Condition进行线程同步协调的时候,一定像我一样先把condition的逻辑语义设计好

  • 将当xxxx时候的表达,设计为condition。
  • 当情况满足condition的时候发出信号signal()通知其他线程;
  • 当情况与condtion正好相反的的时候,使用await阻塞当前线程。

四、put放入数据

其实最重要的就是完成Lock& Condition逻辑设计,剩下的就是填空了,模板如下

通过while循环判断队列当前容量是否达到容量上限,如果达到上限就表示队列满了。队列满了(notFull取反使用await),await阻塞生产线程向队列中继续放入数据。在这里,有小伙伴曾经问过我一个奇葩的问题:多线程持有同一个lock锁,你怎么知道阻塞的是生产线程,而不是消费线程呢? 答:一个线程是生产线程还是消费线程,取决于它的动作(调用什么方法),并没有一个标签给它定义死,调用put方法放入数据的就是生产数据的线程。while/await组合是标准写法,请不要随意创新改成if,否则你会遇到很多诡异的bug。

//队列满了,await阻塞生产线程
while (queue.size() >= queueSize) {
System.out.println(Thread.currentThread().getName() + "等待,因为队列满了" );
notFull.await();
}

向队列中添加一条数据,此时我们可以确定队列是notEmpty,所以使用notEmpty.signal()向生产者发送信号。这里问题又来了:多线程持有同一个lock锁,你怎么知道通知的是消费者线程,而不是生产者线程呢? 答案是我确实不知道,所以在上文中的while (queue.size() >= queueSize)采用的是while,而不是if。即使生产者线程被唤醒了,while判断也会把它await拦住。

//向队列添加一条消息,同时通知消费者有新消息了
queue.add(message);
System.out.println(Thread.currentThread().getName() + "生产" + message );
notEmpty.signal();//通知消费者线程

五、take消费数据

take从队列中取出数据,取出数据之后,队列肯定是notFull ,所以发出notFull.signal信号。当队列空了(notEmpty使用await取反),await同时阻塞消费者线程。

public Object take() throws InterruptedException {
Object retVal = null;
lock.lock();//操作队列先加锁
try {
//队列空了,通知生产线程,消费线程阻塞
while (queue.size() == 0) {
System.out.println("队列已经空了,停止消费!");
notEmpty.await();
}
//队列删除一条消息,同时通知生产者队列有位置了
retVal = queue.get(0);
queue.remove(0);
notFull.signal(); //同时通知生产者队列
} finally {
lock.unlock();
}
return retVal;
}

我相信有了上面put方法的基础,理解take方法中的代码,就非常容易了,这里我就不做过多的说明了。

六、生产消费测试

public static void main(String[] args) {
//为了方便查看测试结果,我们的队列容量设置小一些
MyBlockingQueue queue = new MyBlockingQueue(2); //生产者线程
new Thread(()->{
for(int i = 0;i < 5;i++){
try {
queue.put("msg" + i); //放入5条数据
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start(); //消费者线程
new Thread(()->{
while(true){ //一直消费
try {
System.out.println(Thread.currentThread().getName()
+ "消费数据" + queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}

输出结果如下,满足我们的需求。队列满了,生产者线程Thread-0等待;生产消费互相协调通知,最终数据消费完成,队列空了,消费者线程阻塞。

Thread-0生产msg0
Thread-0生产msg1
Thread-0等待,因为队列满了
Thread-1消费数据msg0
Thread-0生产msg2
Thread-0等待,因为队列满了
Thread-1消费数据msg1
Thread-0生产msg3
Thread-0等待,因为队列满了
Thread-1消费数据msg2
Thread-0生产msg4
Thread-1消费数据msg3
Thread-1消费数据msg4
队列已经空了,停止消费!

欢迎关注我的博客,更多精品知识合集

本文转载注明出处(必须带连接,不能只转文字):字母哥博客 - zimug.com

觉得对您有帮助的话,帮我点赞、分享!您的支持是我不竭的创作动力!。另外,笔者最近一段时间输出了如下的精品内容,期待您的关注。

【java并发编程】Lock & Condition 协调同步生产消费的更多相关文章

  1. Java并发编程的4个同步辅助类

    Java并发编程的4个同步辅助类(CountDownLatch.CyclicBarrier.Semphore.Phaser) @https://www.cnblogs.com/lizhangyong/ ...

  2. Java并发编程:线程的同步

    Java并发编程:线程的同步 */--> code {color: #FF0000} pre.src {background-color: #002b36; color: #839496;} J ...

  3. Java并发编程的4个同步辅助类(CountDownLatch、CyclicBarrier、Semaphore、Phaser)

    我在<JDK1.5引入的concurrent包>中,曾经介绍过CountDownLatch.CyclicBarrier两个类,还给出了CountDownLatch的演示案例.这里再系统总结 ...

  4. Java并发编程的4个同步辅助类(CountDownLatch、CyclicBarrier、Semphore、Phaser)

    我在<jdk1.5引入的concurrent包>中,曾经介绍过CountDownLatch.CyclicBarrier两个类,还给出了CountDownLatch的演示案例.这里再系统总结 ...

  5. Java并发编程:Lock

    Java并发编程:Lock 在上一篇文章中我们讲到了如何使用关键字synchronized来实现同步访问.本文我们继续来探讨这个问题,从Java 5之后,在java.util.concurrent.l ...

  6. Java并发编程,Condition的await和signal等待通知机制

    Condition简介 Object类是Java中所有类的父类, 在线程间实现通信的往往会应用到Object的几个方法: wait(),wait(long timeout),wait(long tim ...

  7. Java并发编程(二)同步

    在多线程的应用中,两个或者两个以上的线程需要共享对同一个数据的存取.如果两个线程存取相同的对象,并且每一个线程都调用了修改该对象的方法,这种情况通常成为竞争条件.  竞争条件最容易理解的例子就是:比如 ...

  8. java 并发编程lock使用详解

    浅谈Synchronized: synchronized是Java的一个关键字,也就是Java语言内置的特性,如果一个代码块被synchronized修饰了,当一个线程获取了对应的锁,执行代码块时,其 ...

  9. Java并发编程(八)同步容器

    为了方便编写出线程安全的程序,Java里面提供了一些线程安全类和并发工具,比如:同步容器.并发容器.阻塞队列.Synchronizer(比如CountDownLatch) 一.为什么会出现同步容器? ...

  10. Java并发编程(六)-- 同步块

    上一节已经讲到,使用Synchronzied代码块可以解决共享对象的竞争问题,其实还有其他的方法也可以避免资源竞争问题,我统称他们为Java同步块.Java 同步块(synchronized bloc ...

随机推荐

  1. 第23章 java线程通信——生产者/消费者模型案例

    第23章 java线程通信--生产者/消费者模型案例 1.案例: package com.rocco; /** * 生产者消费者问题,涉及到几个类 * 第一,这个问题本身就是一个类,即主类 * 第二, ...

  2. python学习文章

    推荐大家看看,这几个涉及到解决问题了 用机器学习算法解决问题(图文+程序) http://www.bida.org.cn/index.php?qa=7 数据的价值是提升业务而不仅仅是用户画像 http ...

  3. iOS学习06C语言结构体

    1.结构体的概述 在C语言中,结构体(struct)指的是一种数据结构,是C语言中构造类型的其中之一. 在实际应用中,我们通常需要由不同类型的数据来构成一个整体,比如学生这个整体可以由姓名.年龄.身高 ...

  4. ajax跨域请求的解决方案

    一直打算改造一下自己传统做网站的形式. 我是.Net程序员,含辛茹苦数年也没混出个什么名堂. 最近微信比较火, 由于现在大环境的影响和以前工作的总结和经验,我打算自己写一个数据,UI松耦合的比较新潮的 ...

  5. centos qt5,PyQt5 installation

    一.SIP http://www.riverbankcomputing.com/software/sip/download   二.Centos6.5 qt 安装 1,centos linux系统必须 ...

  6. &amp;lt;八&amp;gt;阅读&amp;lt;&amp;lt;大话设计模式&amp;gt;&amp;gt;该模型的外观

    Facade模式其实很好理解,被表面的东西展示海报.内部的东西,你不知道(因为我们有一个好包).例如,外部和公司内部制度,5交互系统,此5互.那么第一种就是外部系统和5个系统都进行交互:另外一种就是做 ...

  7. 将一个整数M分成N个整数 要求每个都在区间【minV, maxV】之间

    将一个整数M分成N个整数 要求每个都在区间[minV, maxV]之间,怎么分比较快捷???? 说明: N是>=1且<=9的数,分割的数据只要符合[minV, maxV]区间即可,可以是等 ...

  8. Truck Adblue Emulator For SCANIA

    For sale online Truck Adblue Emulator For SCANIA See the price Where to buy? Truck Adblue Emulator F ...

  9. 本地存储localStorage sessionStorage 以及 session 和cookie的对比和使用

    cookie和session都是用来跟踪浏览器用户身份的会话方式. 1.验证当前服务中继续请求数据时,哪些缓存数据会随着发往服务器? 只有cookie中设置的缓存数据会发送到服务器端 2. 强调几点: ...

  10. [Canvas]RPG游戏雏形 (地图加载,英雄出现并移动)

    源码请点此下载并用浏览器打开index.html观看 图例: 代码: <!DOCTYPE html> <html lang="utf-8"> <met ...