不知道你是否遇到过面试官让你手写生产者消费者代码。别说,前段时间有小伙伴还真的遇到了这种情况。当时是一脸懵逼。
但是,俗话说,从哪里跌倒就要从哪里爬起来。既然这次被问到了,那就回去好好研究一下,争取下一次不再被虐呗。
于是,今天我决定手敲一个生产者消费者模式压压惊。(因为我也不想以后被面试官血虐啊)
生产者消费者模式,其实很简单。无非就是生产者不停的生产数据,消费者不停的消费数据。(这不废话吗,字面意思我也知道啊)
咳咳。其实,我们可以拿水池来举例。
比如,现在要用多个注水管往水池里边注水,那这些注水管就认为是生产者。从水池里边抽水的抽水管就是消费者。水池本身就是一个缓冲区,用于生产者消费者之间的通讯。
好的,跟着我的思路。
既然生产者是生产数据的,那总得定义一个数据类吧(Data)
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内部都需要维护一个阻塞队列,来存储缓冲区的数据。
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; } }消费者:
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(); } } } }测试类,假设有三个注水管和三个出水管(即六个线程)同时运行。等一定时间后,所有注水管停止注水,则当水池空(阻塞队列为空)的时候,出水管也将不再出水。
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。