简介
一般来说，生产者和消费者模式可以实现对任务数量的控制，防止任务过多，导致程序处理不来。
主要思想
- 当队列满时，生产者需要等待队列有空间才能继续往里面放入商品，而在等待的期间内，生产者必须释放对临界资源（即队列）的占用权。因为生产者如果不释放对临界资源的占用权，那么消费者就无法消费队列中的商品，就不会让队列有空间，那么生产者就会一直无限等待下去。
- 因此，一般情况下，当队列满时，会让生产者交出对临界资源的占用权，并进入挂起状态。然后等待消费者消费了商品，然后消费者通知生产者队列有空间了。
- 同样地，当队列空时，消费者也必须等待，等待生产者通知它队列中有商品了。这种互相通信的过程就是线程间的协作。
总结：
- 队列满，生产者等待，消费者消费
- 队列空，消费者等待，生产者生产
案例
假设有生产者可以生产包子，消费者消费包子，这里是实现方式是使用阻塞队列BlockingQueue实现的
1、定义生产者
package productConsumerMode.blockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生产者
*/
public class Producer implements Runnable {
//该字段用于做是否循环的字段 这里使用volatile保证可见性和防止指令重排
private volatile boolean isRunning = true;
//线程休眠时间，防止线程调用速度太快
private final static int SLEEP_TIME = 1000;
//包子阻塞队列，用来存储生产的包子
private BlockingQueue<String> bumQueue;
//包子序号
private static AtomicInteger serial = new AtomicInteger();
/**
* 构造方法
* 参数为：包子队列
*/
public Producer(BlockingQueue<String> bumQueue) {
this.bumQueue = bumQueue;
}
/**
* 重写run方法
*/
@Override
public void run() {
//当isRunning=true的时候，循环进行生产
while (isRunning) {
try {
/**
* boolean offer(E e, long timeout, TimeUnit unit) 若2秒还没有加入，则代表队列阻塞，这时候打印加入队列失败,否则则加入队列
*/
if (!bumQueue.offer("包子" + serial.getAndAdd(1), 2, TimeUnit.SECONDS)) {
System.out.println("队列的数量为：" + bumQueue.size() + "，无法加入队列");
} else {
System.out.println("【生产者】" + Thread.currentThread().getName() + "生产包子" + serial.get());
}
//线程休眠
Thread.sleep(SLEEP_TIME);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void stopProduct() {
isRunning = false;
}
}
2、定义消费者
package productConsumerMode.blockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* 消费者
*/
public class Consumer implements Runnable {
//包子队列
private BlockingQueue<String> bumQueue;
//是否支持消费
private volatile boolean isConsumer=true;
//构造方法
public Consumer(BlockingQueue<String> bumQueue) {
this.bumQueue = bumQueue;
}
@Override
public void run() {
while(isConsumer){
//从队列中获取包子
String bum = bumQueue.poll();
//若包子存在，则消费 若不存在则提示没有包子消费
if (bum != null) {
System.out.println("【消费者】" + Thread.currentThread().getName() + "消费" + bum);
}else{
System.out.println("包子数量剩余"+bumQueue.size()+"，没有包子可以消费");
}
//此处让线程休眠一下，防止消费太快
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void stopConsumer(){
isConsumer=false;
}
}
3、测试
package productConsumerMode.blockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* @Author: WalkerShen
* @DATE: 2022/3/16
* @Description: 使用blockingQueue实现生产者消费者模式
* 案例：生产者生产包子 消费者买包子
**/
public class BlockingQueueTest {
public static void main(String[] args) throws InterruptedException {
//创建一个阻塞队列，容量为10
BlockingQueue<String> bumQueue=new LinkedBlockingQueue<>(10);
//创建Runnable
Producer producer1 = new Producer(bumQueue);
Producer producer2 = new Producer(bumQueue);
Consumer consumer1 = new Consumer(bumQueue);
Consumer consumer2 = new Consumer(bumQueue);
//使用Executors工具执行线程
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.execute(consumer1);
executorService.execute(consumer2);
executorService.execute(producer1);
executorService.execute(producer2);
//让主线程休眠20秒，这里是为了查看执行的结果
Thread.sleep(20*1000);
//停止生产和消费
producer1.stopProduct();
producer2.stopProduct();
consumer1.stopConsumer();
consumer2.stopConsumer();
//最后关闭线程池
executorService.shutdown();
}
}
