面试官让我手写一个生产者消费者模式？

烟雨星空

发布于 2020-06-16 17:47:59

1.3K0

发布于 2020-06-16 17:47:59

文章被收录于专栏：Java进阶指南

不知道你是否遇到过面试官让你手写生产者消费者代码。别说，前段时间有小伙伴还真的遇到了这种情况，当时是一脸懵逼。

但是，俗话说，从哪里跌倒就要从哪里爬起来。既然这次被问到了，那就回去好好研究一下，争取下一次不再被虐呗。

于是，今天我决定手敲一个生产者消费者模式压压惊。（因为我也不想以后被面试官血虐啊）

生产者消费者模式，其实很简单。无非就是生产者不停的生产数据，消费者不停的消费数据。（这不废话吗，字面意思我也知道啊）

咳咳。其实，我们可以拿水池来举例。

比如，现在要用多个注水管往水池里边注水，那这些注水管就认为是生产者。从水池里边抽水的抽水管就是消费者。水池本身就是一个缓冲区，用于生产者消费者之间的通讯。

好的，跟着我的思路。

既然生产者是生产数据的，那总得定义一个数据类吧（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();
    }
}

运行结果如下：