【Java】使用BlockingQueue实现生产者-消费者模式详解编程语言

简介

生产者-消费者模式是一个经典的多线程设计模式。在生产者-消费者模式中。通常有多个生产者线程和多个消费者模式，生产者线程负责提交用户请求，消费者线程负责具体处理生产者提交的任务。两者之间通过共享内存缓冲区进行通信。

生产者-消费者模式的核心组件是共享内存缓冲区，它的作用是生产者与消费者之间的通信桥梁，避免二者之间的直接通信，有效地降低了二者耦合性。生产者不需要知道消费者的存在，消费者也不需要知道生产者的存在。

众所周知，缓存的出现是为了平衡不同的处理速度，比如CPU运算速度要高于硬盘读写速度，如果没有缓存，CPU就会有很多的时间在等待硬盘执行完读写，发挥不了CPU的高性能。

同样，这里的共享内存缓冲区也有这个作用，假如生产者和消费者的速度不匹配，无论是谁要快一些，都可以在这里得到缓解。

生产者-消费者模式的主要角色如下表所示：

实例

这个模式的重点是在共享内存缓冲区为空的时候，生产者需要被唤醒，消费者需要进行等待；当共享内存缓冲区满的时候，消费者需要被唤醒，生产者需要进行等待。这就需要在合适的时候对生产者和消费者休眠和唤醒，也就是notify/wait。这里使用的BlockingQueue内部已经实现了线程的唤醒和休眠，所以在这个代码中看不到notify/wait的出现。

这里假设有若干个生产者进行数据的生成，将生成的数据放到队列中，然后有若干个消费者进行数据的取出，生成这个数的平方。

共享内存缓冲区的数据类型

/** 
 * Created by makersy on 2019 
 */ 
 
/* 
生产者和消费者之间的共享数据 
 */ 
public class Data {
    
 
    int num; 
 
    public Data(int num) {
    
        this.num = num; 
    } 
 
    public Data(String str) {
    
        this.num = Integer.valueOf(str); 
    } 
 
    public int getNum() {
    
        return num; 
    } 
} 
 

生产者

 
/** 
* Created by makersy on 2019 
*/ 
/* 
生产者 
*/ 
public class Producer implements Runnable{
 
//这里的正在运行标记不共享，但是需要用volatile保证可以实时接收到它的更新 
private volatile boolean isRunning = true; 
BlockingQueue<Data> queue; 
//生成的数据,这里要使用static保证在多个线程之间共享 
private static AtomicInteger count = new AtomicInteger(); 
private static final int SLEEPTIME = 1000; 
public Producer(BlockingQueue<Data> queue) {
 
this.queue = queue; 
} 
@Override 
public void run() {
 
Data data = null; 
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{
 
private BlockingQueue<Data> queue; 
private static final int SLEEPTIME = 1000; 
public Consumer(BlockingQueue<Data> queue) {
 
this.queue = queue; 
} 
@Override 
public void run() {
 
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(); 
} 
} 
} 
} 

Main方法（使用生产者和消费者的客户端）

 
/** 
* Created by makersy on 2019 
*/ 
public class Main {
 
public static void main(String[] args) throws InterruptedException {
 
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("结束"); 
} 
}