不知道你是否遇到过面试官让你手写生产者消费者代码。别说，前段时间有小伙伴还真的遇到了这种情况，当时是一脸懵逼。
但是，俗话说，从哪里跌倒就要从哪里爬起来。既然这次被问到了，那就回去好好研究一下，争取下一次不再被虐呗。
于是，今天我决定手敲一个生产者消费者模式压压惊。（因为我也不想以后被面试官血虐啊）
生产者消费者模式，其实很简单。无非就是生产者不停的生产数据，消费者不停的消费数据。（这不废话吗，字面意思我也知道啊）
咳咳。其实，我们可以拿水池来举例。
比如，现在要用多个注水管往水池里边注水，那这些注水管就认为是生产者。从水池里边抽水的抽水管就是消费者。水池本身就是一个缓冲区，用于生产者消费者之间的通讯。
好的，跟着我的思路。
既然生产者是生产数据的，那总得定义一个数据类吧（Data）
代码语言：javascript复制public class Data {
private int id;
private int num;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public int getNum() {
return num;
}
public void setNum(int num) {
this.num = num;
}
public Data(int id, int num) {
this.id = id;
this.num = num;
}
public Data() {
}
}
以上数据，假设注水管每次注水的id和注水容量num(单位是升)都是递增的。并且，单次出水管的出水量和注水管的注水量是一一对应的。
生产者的类Producer和消费者类Consumer内部都需要维护一个阻塞队列，来存储缓冲区的数据。
代码语言：javascript复制public class Producer implements Runnable{
//共享阻塞队列
private BlockingDeque<Data> queue;
//是否还在运行
private volatile boolean isRunning = true;
//id生成器
private static AtomicInteger count = new AtomicInteger();
//生成随机数
private static Random random = new Random();
public Producer(BlockingDeque<Data> queue){
this.queue = queue;
}
@Override
public void run() {
try {
while(isRunning){
//模拟注水耗时
Thread.sleep(random.nextInt(1000));
int num = count.incrementAndGet();
Data data = new Data(num, num);
System.out.println("当前>>注水管:" Thread.currentThread().getName() "注水容量(L):" num);
if(!queue.offer(data,2, TimeUnit.SECONDS)){
System.out.println("注水失败...");
}
}
}catch (Exception e){
e.printStackTrace();
}
}
public void stop(){
isRunning = false;
}
}
消费者：
代码语言：javascript复制public class Consumer implements Runnable{
private BlockingDeque<Data> queue ;
private static Random random = new Random();
public Consumer(BlockingDeque<Data> queue){
this.queue = queue;
}
@Override
public void run() {
while (true){
try {
Data data = queue.take();
//模拟抽水耗时
Thread.sleep(random.nextInt(1000));
if(data != null){
System.out.println("当前<<抽水管:" Thread.currentThread().getName() ",抽取水容量(L):" data.getNum());
}
}catch (Exception e){
e.printStackTrace();
}
}
}
}
测试类，假设有三个注水管和三个出水管（即六个线程）同时运行。等一定时间后，所有注水管停止注水，则当水池空（阻塞队列为空）的时候，出水管也将不再出水。
代码语言：javascript复制public class TestProC {
public static void main(String[] args) throws InterruptedException {
BlockingDeque<Data> queue = new LinkedBlockingDeque<>(10);
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(3000);
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(1000);
service.shutdown();
}
}
运行结果如下：
到最后一次注水20L的时候，所有注水管都停止注水了，但此时水池还没空。于是，所有出水管继续消费水资源，直到最后20L也被消费完。
以上，就是一个典型的生产者消费者模式。
可以看到，这种模式有很多优点：
1）可以解耦消费者和生产者，因为它们是两个不同的类，互相之间不会产生影响。
2）支持并发。生产者只管生产数据就行了，生产完直接把数据丢到缓冲区，而不需要等消费者消费完数据才可以生产下一个数据。否则会造成阻塞，从而影响效率。
3）允许生产者和消费者有不同的处理速度。如，当生产者生产数据比较快的时候，会把消费者还没来得及处理的数据先放到缓冲区。等有空闲的消费者了，再去缓冲区拿去数据。
另外，以上的缓冲区，我们一般会使用阻塞队列。就像上边用的LinkedBlockingDeque。
这样，当队列满的时候，会阻塞生产者继续往队列添加数据，直到有消费者来消费了队列中的数据。当队列空的时候，也会阻塞消费者从队列获取数据，直到有生产者把数据放入到队列中。
阻塞队列最好使用有界队列（代码中指定的容量为10）。因为，如果生产者的速度远远大于消费者时，就会有可能造成队列的元素一直增加，直到内存耗尽。当然，这也需要看实际的业务情况。如果能保证生产者的数量在可控范围内，不会给内存造成压力，用无界队列，也未尝不可。
