当前位置： 首页 > news >正文

java多线程（二一）并发协作生产者消费者设计模式

news 2025/7/3 16:14:47

1.两个线程一个生产者一个消费者

需求情景

  • 两个线程，一个负责生产，一个负责消费，生产者生产一个，消费者消费一个。
    涉及问题

  • 同步问题：如何保证同一资源被多个线程并发访问时的完整性。常用的同步方法是采用标记或加锁机制。

  • wait() / nofity() 方法是基类Object的两个方法，也就意味着所有Java类都会拥有这两个方法，这样，我们就可以为任何对象实现同步机制。

  • wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，放弃锁，使自己处于等待状态，让其他线程执行。

  • notify()方法：当生产者/消费者向缓冲区放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态。\

代码实现（共三个类和一个main方法的测试类）

package com.demo.ProducerConsumer;/*** 资源* @author lixiaoxi**/
public class Resource {/*资源序号*/private int number = 0;/*资源标记*/private boolean flag = false;/*** 生产资源*/public synchronized void create() {if (flag) {//先判断标记是否已经生产了，如果已经生产，等待消费；try {wait();//让生产线程等待} catch (InterruptedException e) {e.printStackTrace();}}number++;//生产一个System.out.println(Thread.currentThread().getName() + "生产者------------" + number);flag = true;//将资源标记为已经生产notify();//唤醒在等待操作资源的线程（队列）}/*** 消费资源*/public synchronized void destroy() {if (!flag) {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread().getName() + "消费者****" + number);flag = false;notify();}
}

package com.demo.ProducerConsumer;/*** 生产者* @author lixiaoxi**/
public class Producer implements Runnable{private Resource resource;public Producer(Resource resource) {this.resource = resource;}@Overridepublic void run() {while (true) {try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}resource.create();}}
}
复制代码
package com.demo.ProducerConsumer;/*** 消费者* @author lixiaoxi**/
public class Consumer implements Runnable{private Resource resource;public Consumer(Resource resource) {this.resource = resource;}@Overridepublic void run() {while (true) {try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}resource.destroy();}}
}package com.demo.ProducerConsumer;public class ProducerConsumerTest {public static void main(String args[]) {Resource resource = new Resource();new Thread(new Producer(resource)).start();//生产者线程new Thread(new Consumer(resource)).start();//消费者线程}
}

在这里插入图片描述

二、多个线程，多个生产者和多个消费者的问题

需求情景

四个线程，两个个负责生产，两个个负责消费，生产者生产一个，消费者消费一个。
涉及问题

notifyAll()方法：当生产者/消费者向缓冲区放入/取出一个产品时，向其他等待的所有线程发出可执行的通知，同时放弃锁，使自己处于等待状态。

测试代码如下

package com.demo.ProducerConsumer;public class ProducerConsumerTest {public static void main(String args[]) {Resource resource = new Resource();new Thread(new Producer(resource)).start();//生产者线程new Thread(new Producer(resource)).start();//生产者线程new Thread(new Consumer(resource)).start();//消费者线程new Thread(new Consumer(resource)).start();//消费者线程}
}

在这里插入图片描述
通过以上打印结果发现问题

147生产了一次，消费了两次。
169生产了，而没有消费。
原因分析

当两个线程同时操作生产者生产或者消费者消费时，如果有生产者或消费者的两个线程都wait()时，再次notify()，由于其中一个线程已经改变了标记而另外一个线程再次往下直接执行的时候没有判断标记而导致的。
if判断标记，只有一次，会导致不该运行的线程运行了。出现了数据错误的情况。
解决方案

while判断标记，解决了线程获取执行权后，是否要运行！也就是每次wait()后再notify()时先再次判断标记。

代码改进（Resource中的 if -> while）

package com.demo.ProducerConsumer;/*** 资源* @author lixiaoxi**/
public class Resource {/*资源序号*/private int number = 0;/*资源标记*/private boolean flag = false;/*** 生产资源*/public synchronized void create() {while (flag) {//先判断标记是否已经生产了，如果已经生产，等待消费；try {wait();//让生产线程等待} catch (InterruptedException e) {e.printStackTrace();}}number++;//生产一个System.out.println(Thread.currentThread().getName() + "生产者------------" + number);flag = true;//将资源标记为已经生产notify();//唤醒在等待操作资源的线程（队列）}/*** 消费资源*/public synchronized void destroy() {while (!flag) {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread().getName() + "消费者****" + number);flag = false;notify();}
}

在这里插入图片描述
再次发现问题

  • 打印到某个值比如生产完187，程序运行卡死了，好像锁死了一样。
    原因分析

  • notify：只能唤醒一个线程，如果本方唤醒了本方，没有意义。而且while判断标记+notify会导致”死锁”。
    解决方案

  • notifyAll解决了本方线程一定会唤醒对方线程的问题。

最后代码改进（Resource中的 notify() -> notifyAll()）

package com.demo.ProducerConsumer;/*** 资源* @author lixiaoxi**/
public class Resource {/*资源序号*/private int number = 0;/*资源标记*/private boolean flag = false;/*** 生产资源*/public synchronized void create() {while (flag) {//先判断标记是否已经生产了，如果已经生产，等待消费；try {wait();//让生产线程等待} catch (InterruptedException e) {e.printStackTrace();}}number++;//生产一个System.out.println(Thread.currentThread().getName() + "生产者------------" + number);flag = true;//将资源标记为已经生产notifyAll();//唤醒在等待操作资源的线程（队列）}/*** 消费资源*/public synchronized void destroy() {while (!flag) {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread().getName() + "消费者****" + number);flag = false;notifyAll();}
}

在这里插入图片描述
以上就大功告成了，没有任何问题。
再来梳理一下整个流程。按照示例，生产者消费者交替运行，每次生产后都有对应的消费者，测试类创建实例，如果是生产者先运行，进入run()方法，进入create()方法，flag默认为false，number+1，生产者生产一个产品，flag置为true，同时调用notifyAll()方法，唤醒所有正在等待的线程，接下来如果还是生产者运行呢？这是flag为true，进入while循环，执行wait()方法，接下来如果是消费者运行的话，调用destroy()方法，这时flag为true，消费者购买了一次产品，随即将flag置为false，并唤醒所有正在等待的线程。这就是一次完整的多生产者对应多消费者的问题。

三、使用Lock和Condition来解决生产者消费者问题

上面的代码有一个问题，就是我们为了避免所有的线程都处于等待的状态，使用了notifyAll方法来唤醒所有的线程，即notifyAll唤醒的是自己方和对方线程。如果我需要只是唤醒对方的线程，比如：生产者只能唤醒消费者的线程，消费者只能唤醒生产者的线程。

在jdk1.5当中为我们提供了多线程的升级解决方案：

  1. 将同步synchronized替换成了Lock操作。

  2. 将Object中的wait，notify，notifyAll方法替换成了Condition对象。

  3. 可以只唤醒对方的线程。

完整代码：

package com.demo.ProducerConsumer;import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;/*** 资源* @author lixiaoxi**/
public class Resource1 {/*资源序号*/private int number = 0;/*资源标记*/private boolean flag = false;private Lock lock = new ReentrantLock();//使用lock建立生产者的condition对象private Condition condition_pro = lock.newCondition(); //使用lock建立消费者的condition对象private Condition condition_con = lock.newCondition(); /*** 生产资源*/public void create() throws InterruptedException {try{lock.lock();//先判断标记是否已经生产了，如果已经生产，等待消费while(flag){//生产者等待condition_pro.await();}//生产一个number++;System.out.println(Thread.currentThread().getName() + "生产者------------" + number);//将资源标记为已经生产flag = true;//生产者生产完毕后，唤醒消费者的线程（注意这里不是signalAll)condition_con.signal();}finally{lock.unlock();}}/*** 消费资源*/public void destroy() throws InterruptedException{try{lock.lock();//先判断标记是否已经消费了，如果已经消费，等待生产while(!flag){//消费者等待condition_con.await();}System.out.println(Thread.currentThread().getName() + "消费者****" + number);//将资源标记为已经消费flag = false;//消费者消费完毕后，唤醒生产者的线程condition_pro.signal();}finally{lock.unlock();}}
}package com.demo.ProducerConsumer;/*** 生产者* @author lixiaoxi**/
public class Producer1 implements Runnable{private Resource1 resource;public Producer1(Resource1 resource) {this.resource = resource;}@Overridepublic void run() {while (true) {try {Thread.sleep(10);resource.create();} catch (InterruptedException e) {e.printStackTrace();}}}}package com.demo.ProducerConsumer;/*** 消费者* @author lixiaoxi**/
public class Consumer1 implements Runnable{private Resource1 resource;public Consumer1(Resource1 resource) {this.resource = resource;}@Overridepublic void run() {while (true) {try {Thread.sleep(10);resource.destroy();} catch (InterruptedException e) {e.printStackTrace();}}}}package com.demo.ProducerConsumer;public class ProducerConsumerTest1 {public static void main(String args[]) {Resource1 resource = new Resource1();new Thread(new Producer1(resource)).start();//生产者线程new Thread(new Producer1(resource)).start();//生产者线程new Thread(new Consumer1(resource)).start();//消费者线程new Thread(new Consumer1(resource)).start();//消费者线程}
}

在这里插入图片描述

四、总结

1、如果生产者、消费者都是1个，那么flag标记可以用if判断。这里有多个，必须用while判断。

2、在while判断的同时，notify函数可能唤醒本类线程(如一个消费者唤醒另一个消费者)，这会导致所有消费者忙等待，程序无法继续往下执行。使用notifyAll函数代替notify可以解决这个问题，notifyAll可以保证非本类线程被唤醒(消费者线程能唤醒生产者线程，反之也可以)，解决了忙等待问题。

小心假死

生产者/消费者模型最终达到的目的是平衡生产者和消费者的处理能力，达到这个目的的过程中，并不要求只有一个生产者和一个消费者。可以多个生产者对应多个消费者，可以一个生产者对应一个消费者，可以多个生产者对应一个消费者。
假死就发生在上面三种场景下。假死指的是全部线程都进入了WAITING状态，那么程序就不再执行任何业务功能了，整个项目呈现停滞状态。
比方说有生产者A和生产者B，缓冲区由于空了，消费者处于WAITING。生产者B处于WAITING，生产者A被消费者通知生产，生产者A生产出来的产品本应该通知消费者，结果通知了生产者B，生产者B被唤醒，发现缓冲区满了，于是继续WAITING。至此，两个生产者线程处于WAITING，消费者处于WAITING，系统假死。上面的分析可以看出，假死出现的原因是因为notify的是同类，所以非单生产者/单消费者的场景，可以采取两种方法解决这个问题：

（1）synchronized用notifyAll()唤醒所有线程、ReentrantLock用signalAll()唤醒所有线程。

（2）用ReentrantLock定义两个Condition，一个表示生产者的Condition，一个表示消费者的Condition，唤醒的时候调用相应的Condition的signal()方法就可以了。

http://www.lryc.cn/news/33837.html

相关文章：

  • Win YAPI + Jenkins 实现接口自动化测试
  • 【计算机视觉 自然语言处理】什么是多模态？
  • 2023百度面试真题
  • MAC(m1)-VMWare Fushion安装Windows11
  • HTML与CSS简介
  • 基于Java开发幼儿园管理系统项目教程(附源码)
  • 第一次运行vue遇到的问题
  • Clickhouse数据去重
  • 精讲typescript从入门到入土
  • typora-beta-0.11.18版本又提示过期的解决方案
  • WebUI自动化测试框架搭建（二十）-优化：测试对象无法连接或出现异常时，请更新本文作为测试对象
  • 【FATE联邦学习】standalone版Fateboard修改配置
  • 分享一个应急响应web日志：access.log文件分析小工具
  • windows注册服务非常实用
  • 蓝桥dfs专题
  • [ 网络 ] 应用层协议——HTTPS协议原理
  • http协议如何操作
  • ESP Insights 系列文章
  • 如何提高爬虫工作效率
  • React结合Drag API实现拖拽示例详解
  • 【华为OD机试java、python、c++、jsNode】新学校选址（100%通过+复盘思路）
  • Nacos配置中心，分组配置参考，以及python、go、bash客户端连接获取
  • node-red中有关用户登录，鉴权，权限控制的流程解析
  • MQTT协议-使用CONNECT报文连接阿里云
  • 每日学术速递3.8
  • 测牛学堂：软件测试之接口测试理论基础总结
  • 基于土壤数据与机器学习算法的农作物推荐算法代码实现
  • python中html必备基础知识
  • 【专项训练】前言：刻意练习，不断的过遍数才是王道
  • 【Leetcode】反转链表 合并链表 相交链表 链表的回文结构