【Java】使用BlockingQueue实现生产者-消费者模式详解编程语言
2023-06-13 09:11:51 时间
简介
实例
生产者-消费者模式是一个经典的多线程设计模式。在生产者-消费者模式中。通常有多个生产者线程和多个消费者模式,生产者线程负责提交用户请求,消费者线程负责具体处理生产者提交的任务。两者之间通过共享内存缓冲区进行通信。
生产者-消费者模式的核心组件是共享内存缓冲区,它的作用是生产者与消费者之间的通信桥梁,避免二者之间的直接通信,有效地降低了二者耦合性。生产者不需要知道消费者的存在,消费者也不需要知道生产者的存在。
众所周知,缓存的出现是为了平衡不同的处理速度,比如CPU运算速度要高于硬盘读写速度,如果没有缓存,CPU就会有很多的时间在等待硬盘执行完读写,发挥不了CPU的高性能。
同样,这里的共享内存缓冲区也有这个作用,假如生产者和消费者的速度不匹配,无论是谁要快一些,都可以在这里得到缓解。
生产者-消费者模式的主要角色如下表所示:
实例
这个模式的重点是在共享内存缓冲区为空的时候,生产者需要被唤醒,消费者需要进行等待;当共享内存缓冲区满的时候,消费者需要被唤醒,生产者需要进行等待。这就需要在合适的时候对生产者和消费者休眠和唤醒,也就是notify/wait。这里使用的BlockingQueue内部已经实现了线程的唤醒和休眠,所以在这个代码中看不到notify/wait的出现。
这里假设有若干个生产者进行数据的生成,将生成的数据放到队列中,然后有若干个消费者进行数据的取出,生成这个数的平方。
共享内存缓冲区的数据类型/** * Created by makersy on 2019 生产者和消费者之间的共享数据 public class Data {消费者
//这里的正在运行标记不共享,但是需要用volatile保证可以实时接收到它的更新 private volatile boolean isRunning = true; BlockingQueue Data queue; //生成的数据,这里要使用static保证在多个线程之间共享 private static AtomicInteger count = new AtomicInteger(); private static final int SLEEPTIME = 1000;
System.out.println("Producer Thread id: " + Thread.currentThread().getId() + " started!"); Random r = new Random(); try { while (isRunning) { Thread.sleep(r.nextInt(SLEEPTIME)); data = new Data(count.incrementAndGet()); System.out.println(data.getNum() + " is put into bq"); if (queue.offer(data, 2, TimeUnit.SECONDS)) { //生产者向bq中添加数据 System.out.println("producer " + Thread.currentThread().getId() + " put data : " + data.getNum()); } else { System.out.println("producer put data failed!"); } catch (InterruptedException e) { e.printStackTrace(); Thread.interrupted(); //停止线程 public void stop() { isRunning = false;
* Created by makersy on 2019 public class Consumer implements Runnable{Main方法(使用生产者和消费者的客户端)
System.out.println("Consumer Thread id: " + Thread.currentThread().getId() + " started!"); Random r = new Random(); while (true) { try { Data data = queue.take(); if (data != null) { //取出数据成功,输出计算平方值 System.out.println(Thread.currentThread().getId() + "计算平方:" + MessageFormat.format("{0} * {1} = {2}", data.getNum(), data.getNum(), data.getNum() * data.getNum())); Thread.sleep(r.nextInt(SLEEPTIME)); //随机睡眠一定时间,模拟任务执行 } catch (InterruptedException e) { e.printStackTrace(); Thread.interrupted();
* Created by makersy on 2019 public class Main {
BlockingQueue Data queue = new LinkedBlockingQueue (10); //新建3个生产者线程和3个消费者线程 Producer producer1 = new Producer(queue); Producer producer2 = new Producer(queue); Producer producer3 = new Producer(queue); Consumer consumer1 = new Consumer(queue); Consumer consumer2 = new Consumer(queue); Consumer consumer3 = new Consumer(queue); //线程池 ExecutorService service = Executors.newCachedThreadPool(); service.execute(producer1); service.execute(producer2); service.execute(producer3); service.execute(consumer1); service.execute(consumer2); service.execute(consumer3); Thread.sleep(10 * 1000); //执行生产者和消费者 producer1.stop(); producer2.stop(); producer3.stop(); //停止生产任务 Thread.sleep(3 * 1000); //等待消费者处理完队列中的任务 service.shutdown(); System.out.println("结束");
原创文章,作者:ItWorker,如若转载,请注明出处:https://blog.ytso.com/19369.html
cjavaxml相关文章
- java单例模式_Java单例模式
- java的三种工厂模式「建议收藏」
- Java实现单例模式(懒汉式和饿汉式)
- 浅析Java多态_JAVA多态
- Java截取字符串方法_java通过split截取字符串
- java prototype是什么,Java设计模式之原型模式(Prototype模式)介绍
- c 线程安全的单例模式-c多线程并发处理方式_Java多线程面试题:线程锁+线程池+线程同步等
- java JNI: C 语言调用 Java 方法示例详解编程语言
- java策略模式详解编程语言
- Java学习笔记之六java三种循环(for,while,do……while)的使用方法及区别详解编程语言
- Java学习笔记之四java进制转化详解编程语言
- Java学习笔记之二java标识符命名规范详解编程语言
- java I/O之装饰者模式详解编程语言
- Java程序员必备知识,《JAVA编程思想》包和访问权限详解编程语言
- Linux下Java开发:给初学者的指引(linux下java编程)
- Java问题-java进程占用内存过高,排查原因详解编程语言
- Linux上部署Java项目实践指南(linux部署java项目)
- Java无法连接Redis服务器(java连不上redis)
- 使用Java轻松读取Redis数据 – 一步一步指南!(java读取redis)
- Linux下Java命令:简介与基本用法(linux下java命令)
- Java配置Oracle实现稳定的跨平台数据库连接(java配置oracle)
- Java实现Redis计数器功能(redis计数 java)