消费者

package consumer;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;public class Consumer implements Runnable {/** 用util.concurrent.BlockingQueue沟通生产者和消费者的桥梁*/BlockingQueue<String> queue;String id;@SuppressWarnings("unused")private volatile boolean      isRunning               = true;public Consumer(BlockingQueue<String> queue, String id) {this.queue = queue;this.id = id;}public void stop() {isRunning = false;}@Overridepublic void run() {System.out.println("Thread: " + id + " Consumer thread is running...");boolean isRunning = true;try {while (isRunning) {System.out.println("Thread: " + id + " fetch data from linkedQueue..." + " queue size: " + queue.size());/** 从队列里取出一个元素，2秒超时，如果两秒之后还没有东西可以取，则poll返回null*/String data = queue.poll(2, TimeUnit.SECONDS);if (null != data) {System.out.println("Thread: " + id + " has consumed one data from queue: " + data+ "   Queue sise: " + queue.size());// simulate data consumptionThread.sleep(1000);} else {isRunning = false;// 消费者准备退出System.out.println("Thread: " + id + " Consumer read queue timeout");}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("Thread: " + id + " consumer thread ends");}}}

生产者

package consumer;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;public class Producer implements Runnable {BlockingQueue<String> queue;String id;public Producer(BlockingQueue<String> queue, String id) {this.queue = queue;this.id = id;}@Overridepublic void run() {String data = null;try {while (isRunning) {System.out.println("PRODUCER: " + id + " is running");Thread.sleep(100);data = "data:" + count.incrementAndGet();System.out.println("Thread: " + id + " procedued data into queue: " + data + " ...");if (!queue.offer(data, 2, TimeUnit.SECONDS)) {System.out.println("failed to put data into queue: " + data);}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("Thread: " + id + " quit from producer thread");}}public void stop() {isRunning = false;}private volatile boolean      isRunning               = true;private static AtomicInteger  count                   = new AtomicInteger();}

测试代码

package consumer;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;public class ConsumerProducerTest {public static void main(String[] args) throws InterruptedException {BlockingQueue<String> queue = new LinkedBlockingQueue<String>(15);Producer producer1 = new Producer(queue, "PROD1");Producer producer2 = new Producer(queue, "PROD2");Producer producer3 = new Producer(queue, "PROD3");Consumer consumer1 = new Consumer(queue, "CONSUMER1");Consumer consumer2 = new Consumer(queue, "CONSUMER2");ExecutorService service = Executors.newCachedThreadPool();service.execute(producer1);service.execute(producer2);service.execute(producer3);service.execute(consumer1);service.execute(consumer2);Thread.sleep(3 * 1000);producer1.stop(); // 一定要先关闭生产者producer2.stop();producer3.stop();consumer1.stop();consumer2.stop();Thread.sleep(2000);service.shutdown();}
}

一个使用Java BlockingQueue实现的生产者和消费者相关推荐

详细讲解如何使用Java连接Kafka构建生产者和消费者（带测试样例）
1 缘起学习消息队列的过程中,先补习了RabbitMQ相关知识, 接着又重温了Kafka相关的知识, 发现,我并没有积累Java原生操作Kafka的文章, 只使用SpringBoot集成过Kafka ...
Java基础进阶多线程-生产者和消费者模式
1.什么是"生产者和消费者模式"? 生产线程负责生产,消费线程负责消费生产线程和消费线程要达到均衡这是一种特殊的业务需求,在这种特殊的情况下需要使用wait方法和notify方 ...
基于java多线程来实现生产者和消费者的实例
声明:本实例是在网上看到,做了很小的修改.所以感谢之前的作者.只是一时忘了哪儿看到,没法加入链接,向原作者道歉,以示尊重.抱歉-^)... 同步栈: 1 class SycnStack { 2 pri ...
Java多线程2.3.生产者与消费者之间的关系2
生产者与消费者之间的关系 1.线程间通信问题描述图 2.线程的状态转换图及常见执行情况 3.等待唤醒机制思路图解 4.线程的生命周期
java 多线程经典例子——生产者与消费者的问题
产品名称类: public class Product {//产品名称private String name;public String getName() {return name;}public ...
C#多线程学习(三) 生产者和消费者
C#多线程学习(三) 生产者和消费者原文链接:http://kb.cnblogs.com/page/42530/ 本系列文章导航 C#多线程学习(一) 多线程的相关概念 C#多线程学习(二) 如何操 ...
生产者与消费者问题------legend050709
生产者与消费者问题: (一)基础: (1.0)生产者消费者的背景 1>为了平衡生产者和消费者的处理能力,起到一个数据缓存的作用,同时也达到了一个解耦的作用在多线程开发中,如果生产者生产数据 ...
【JAVA多线程】如何解决一个生产者与消费者问题
如何解决一个生产者与消费者问题生产者与消费者问题是多线程同步的一个经典问题.生产者和消费者同时使用一块缓冲区,生产者生产商品放入缓冲区,消费者从缓冲区中取出商品.我们需要保证的是,当缓冲区满时,生产 ...
Java中生产者与消费者问题的演变
想要了解更多关于Java生产者消费者问题的演变吗?那就看看这篇文章吧,我们分别用旧方法和新方法来处理这个问题. 生产者消费者问题是一个典型的多进程同步问题. 对于大多数人来说,这个问题可能是我们在学校 ...

一个使用Java BlockingQueue实现的生产者和消费者

消费者

生产者

测试代码

一个使用Java BlockingQueue实现的生产者和消费者相关推荐

最新文章

热门文章