【手撕代码】生产者消费者问题

小咪咪 2022-02-16 01:13 197阅读 0赞

本文转自：[https://www.jianshu.com/p/3f0cd7af370d][https_www.jianshu.com_p_3f0cd7af370d]

*  在现场面试的过程中，经常遇到手写生产者和消费者的问题。这里做个总结：

这绝对是属于重点了，不管是考察对于该重要模型的理解还是考察代码能力，这都是一道很好的考题，所以很有必要的，我们先来回顾一下什么是生产者-消费者问题。

## 一、问题简单回顾 ##

![Image 1][]

**生产者消费者问题**（英语：Producer-Consumer problem），也称**有限缓冲问题**（英语：Bounded-buffer problem），是一个[多线程][Link 1][同步][Link 2]问题的经典案例。该问题描述了共享固定大小[缓冲区][Link 3]的两个线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。**该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。**

*  **注意：** 生产者-消费者模式中的内存缓存区的主要功能是数据在多线程间的共享，此外，通过该缓冲区，可以缓解生产者和消费者的性能差；

## 二、几种实现方式 ##

上面说到该问题的关键是：**如何保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区空时消耗数据；**解决思路可以简单概括为：

*  生产者持续生产，直到缓冲区满，**满时阻塞**；缓冲区不满后，继续生产；
 *  消费者持续消费，直到缓冲区空，**空时阻塞**；缓冲区不空后，继续消费；
 *  生产者和消费者都可以有多个；

那么在 Java 语言中，能达到上述要求的，自然而然的就会有如下的几种写法，但是问题的核心都是能够让消费者和生产者在各自满足条件需要阻塞时能够起到正确的作用：

> **`1、wait()`/`notify()`方式；**
> 
> **`2、await()`/`signal()`方式；**
> 
> **`3、BlockingQueue`阻塞队列方式；**
> 
> **`4、PipedInputStream`/`PipedOutputStream`方式；**

手写代码，我们着重掌握上面对应的第一种和第三种写法就足够了；

### 2.1 wait / notify 方式实现 ###

*  **生产者 **

public class Producer implements Runnable {
    
        // 为 true 时，生产者生产，为 false 时停止生产
        private volatile boolean needRpoduce = true;
    
        // 公共资源
        private final Vector sharedQueue;
    
        // 公共资源的最大数量
        private final int SIZE;
    
        // 生产数据
        private static AtomicInteger count = new AtomicInteger();
    
        public Producer(Vector sharedQueue, int SIZE){
            this.sharedQueue = sharedQueue;
            this.SIZE = SIZE;
        }
    
    
        @Override
        public void run() {
            int data;
            Random random = new Random();
    
            System.out.println("start producer id = " + Thread.currentThread().getId());
            try{
                while(needRpoduce){
                    // 模拟延迟
                    Thread.sleep(random.nextInt(1000));
                    // 当队列已经满了的时候，就阻塞等待
                    while(sharedQueue.size() == SIZE){
                        synchronized(sharedQueue){
                            System.out.println("Queue is full, producer " + Thread.currentThread().getId() + " is waiting, size：" + sharedQueue.size());
                            sharedQueue.wait();
                        }
                    }
    
                    // 阻塞队列不满时就继续生产
                    synchronized(sharedQueue){
                        // 生产数据
                        data = count.incrementAndGet();
                        sharedQueue.add(data);
                        System.out.println("producer create data:" + data + ", size：" + sharedQueue.size());
                        sharedQueue.notifyAll();
                    }
    
                }
            }catch (InterruptedException e){
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
        }
    
        // 停止生产
        public void stop(){
            needRpoduce = false;
        }
    }

> 有了上面的提到的解决思路，应该很容易实现，但是这里主要提一下一些细节和重点：
> 
>  *  **创造数据**：生产者-消费者解决的问题就是数据在多线程间的共享，所以我们首要关心的问题就应该是数据，我们这里采用的是使用一个 **`AtomicInteger`**类来为我们创造数据，使用它的好处是该类是一个保证原子操作的类，我们使用其中的**`incrementAndGet()`**方法不仅能够保证线程安全，还可以达到一个计数的效果，所以是一个既简单又实用的选择，当然也可以使用其他的数据来代替，**这里注意的是要保证该类在内存中只存在一份，使用`static`修饰**；
>  *  **内存缓冲区**：要保证在多线程环境下内存缓冲区的安全，所以我们考虑使用简单的**`Vector`**类来作为我们的内存缓冲区，并且使用**`final`**修饰保证内存缓冲区的唯一，然后的话我们需要判断队列是否满，需要手动添加一个标识缓冲区大小的变量**`SIZE`**，注意也是`final`修饰；
>  *  **模拟延迟**：这里主要模拟的是一个网络延迟，然后使用`Random()`类的`nextInt()`方法来随机选取一个1000范围内的值来模拟网络环境中的延迟；
>  *  **停止方法**：首先需要知道在`Thread`类中有一个弃用的`stop()`方法，我们自己增加一个标志位`needProduce`来完成我们自己的`stop()`功能，**需要注意的是使用`volatile`来修饰**，保证该标志位的可见性；
>  *  **错误处理**：当捕获到错误时，我们应该使用`Thread`类中的`interrupted()`方法来终止当前的进程；
>  *  **消息提示**：我们主要是要在控制台输出该生产者的信息，包括当前队列的状态，大小，当前线程的生产者信息等，**注意的是信息格式的统一(后面的消费者同样的)**；

*  **消费者**

public class Consumer implements Runnable {
    
        // 公共资源
        private final Vector sharedQueue;
    
        public Consumer(Vector sharedQueue){
            this.sharedQueue = sharedQueue;
        }
    
        @Override
        public void run() {
    
            Random random = new Random();
            System.out.println("start consumer id = " + Thread.currentThread().getId());
            try{
                while(true){
                    // 模拟延迟
                    Thread.sleep(1000);
                    // 当队列为空时，阻塞等待生产者生产新的数据
                    while(sharedQueue.isEmpty()){
                        synchronized(sharedQueue){
                            System.out.println("Queue is empty, consumer " + Thread.currentThread().getId() + " is waiting, size：" + sharedQueue.size());
                            sharedQueue.wait();
                        }
                    }
                    // 队列不为空时，直接消费
                    synchronized(sharedQueue){
                        System.out.println("consumer consume data：" + sharedQueue.remove(0) + ", size：" + sharedQueue.size());
                        sharedQueue.notifyAll();
                    }
                }
            }catch(InterruptedException e){
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
        }
    }

跟生产者相同的，你需要注意**内存缓冲区**/ **模拟延迟**/ **错误处理**/ **消息提示**这些方面的细节问题，总体来说消费者就是持续不断的消费，也比较容易实现。

*  **主线程代码**

有了我们的消费者和生产者代码，我们需要来验证一下它们的正确性，照常理来说我们直接创建一些消费者和生产者的线程让它们执行就可以了啊，但是为了“加分”考虑呢，我们还是使用线程池吧..也不是特别复杂：

public class Test {
    
        public static void main(String[] args) throws InterruptedException {
    
            // 1. 建立内存缓冲区
            Vector sharedQueue = new Vector();
            int size = 4;
    
            // 2. 构建线程池和线程
            ExecutorService service = Executors.newCachedThreadPool();
            Producer producer1 = new Producer(sharedQueue,size);
            Producer producer2 = new Producer(sharedQueue,size);
            Producer producer3 = new Producer(sharedQueue,size);
            Consumer consumer1 = new Consumer(sharedQueue);
            Consumer consumer2 = new Consumer(sharedQueue);
            Consumer consumer3 = new Consumer(sharedQueue);
            service.execute(producer1);
            service.execute(producer2);
            service.execute(producer3);
            service.execute(consumer1);
            service.execute(consumer2);
            service.execute(consumer3);
    
            // 3. 睡一会儿，然后尝试停止生产者（结束循环）
            Thread.sleep(10 * 1000);
            producer1.stop();
            producer2.stop();
            producer3.stop();
    
            // 4. 再睡一会儿，关闭线程池
            Thread.sleep(3000);
    
            // 5. shutdown() 等待任务执行完才中断线程(消费者线程未停止，所以其实程序是无法结束的)
            service.shutdown();
        }
    }

*  ** 运行结果**

start producer id = 13
    start consumer id = 16
    start producer id = 12
    start producer id = 14
    start consumer id = 17
    start consumer id = 15
    producer create data:1, size：1
    consumer consume data：1, size：0
    producer create data:2, size：1
    producer create data:3, size：2
    consumer consume data：2, size：1
    consumer consume data：3, size：0
    producer create data:4, size：1
    consumer consume data：4, size：0
    Queue is empty, consumer 16 is waiting, size：0
    producer create data:5, size：1
    producer create data:6, size：2
    consumer consume data：5, size：1
    consumer consume data：6, size：0
    producer create data:7, size：1
    producer create data:8, size：2
    consumer consume data：7, size：1
    consumer consume data：8, size：0
    producer create data:9, size：1
    Queue is empty, consumer 16 is waiting, size：1
    producer create data:10, size：2
    consumer consume data：9, size：1
    producer create data:11, size：2
    consumer consume data：10, size：1
    consumer consume data：11, size：0
    producer create data:12, size：1
    producer create data:13, size：2
    consumer consume data：12, size：1
    consumer consume data：13, size：0
    producer create data:14, size：1
    consumer consume data：14, size：0
    producer create data:15, size：1
    producer create data:16, size：2
    producer create data:17, size：3
    producer create data:18, size：4
    consumer consume data：15, size：3
    consumer consume data：16, size：2
    consumer consume data：17, size：1
    producer create data:19, size：2
    consumer consume data：18, size：1
    consumer consume data：19, size：0
    producer create data:20, size：1
    consumer consume data：20, size：0
    producer create data:21, size：1
    consumer consume data：21, size：0
    Queue is empty, consumer 17 is waiting, size：0
    producer create data:22, size：1
    consumer consume data：22, size：0
    producer create data:23, size：1
    producer create data:24, size：2
    consumer consume data：23, size：1
    consumer consume data：24, size：0
    producer create data:25, size：1
    consumer consume data：25, size：0
    producer create data:26, size：1
    producer create data:27, size：2
    consumer consume data：26, size：1
    consumer consume data：27, size：0
    Queue is empty, consumer 17 is waiting, size：0
    producer create data:28, size：1
    consumer consume data：28, size：0
    producer create data:29, size：1
    consumer consume data：29, size：0
    producer create data:30, size：1
    consumer consume data：30, size：0
    Queue is empty, consumer 17 is waiting, size：0
    Queue is empty, consumer 15 is waiting, size：0
    Queue is empty, consumer 16 is waiting, size：0

### 2.2 BlockingQueue 阻塞队列实现方式 ###

任何有效的生产者-消费者问题解决方案都是通过控制生产者put()方法（生产资源）和消费者take()方法（消费资源）的调用来实现的，一旦你实现了对方法的阻塞控制，那么你将解决该问题。

Java通过**`BlockingQueue`**提供了开箱即用的支持来控制这些方法的调用（一个线程创建资源，另一个消费资源）。**`java.util.concurrent`**包下的**`BlockingQueue`**接口是一个线程安全的可用于存取对象的队列。

![879896-20160602152903211-273313581.png][]

BlockingQueue是一种数据结构，支持一个线程往里存资源，另一个线程从里取资源。这正是解决生产者消费者问题所需要的，那么让我们开始解决该问题吧。

*  **Producer**

public class Producer implements Runnable{
    
        private volatile boolean needProduce = true;
        private BlockingQueue<Integer> queue;                       // 内存缓冲区
        private static AtomicInteger count = new AtomicInteger();   // 总数，原子操作
    
        public Producer(BlockingQueue<Integer> queue){
            this.queue = queue;
        }
    
        @Override
        public void run() {
            int data;
            Random random = new Random();
            System.out.println("start producer id = " + Thread.currentThread().getId());
            try{
                while(needProduce){
                    // 模拟延迟
                    Thread.sleep(random.nextInt(1000));
    
                    // 往阻塞队列中添加数据
                    data = count.incrementAndGet();
                    System.out.println("producer " + Thread.currentThread().getId() + " create data：" + data + ", size：" + queue.size());
                    if(!queue.offer(data, 2, TimeUnit.SECONDS)){
                        System.err.println("failed to put data：" + data);
                    }
                }
            }catch (InterruptedException e){
                e.printStackTrace();
                Thread.currentThread().interrupted();
            }
        }
    
        // 停止生产者线程
        public void stop(){
            needProduce = false;
        }
    }

跟上面一种方式没有很大的差别，倒是代码更加简单通透，不过需要注意的是对**阻塞队列添加失败的错误处理。**

*  **Consumer**

public class Consumer implements Runnable {
    
        private BlockingQueue<Integer> queue;     // 内存缓冲区
    
        public Consumer(BlockingQueue<Integer> queue){
            this.queue = queue;
        }
    
        @Override
        public void run() {
            int data;
            Random random = new Random();
            System.out.println("start consumer id = " + Thread.currentThread().getId());
    
            try{
                while(true){
                    // 模拟延迟
                    Thread.sleep(random.nextInt(1000));
    
                    // 从阻塞队列中消费数据
                    if(!queue.isEmpty()){
                        data = queue.take();
                        System.out.println("consumer " + Thread.currentThread().getId() + " consume data：" + data + ", size：" + queue.size());
                    }else{
                        System.out.println("Queue is empty, consumer " + Thread.currentThread().getId() + " is waiting, size：" + queue.size());
                    }
                }
            }catch (InterruptedException e){
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
        }
    }

*  **主线程代码**

public class Test {
    
        public static void main(String[] args) throws InterruptedException {
    
            // 1. 构建内存缓冲区
            BlockingQueue<Integer> queue = new LinkedBlockingDeque<>();
    
            // 2. 构建线程池和线程
            ExecutorService service = Executors.newCachedThreadPool();
            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);
            service.execute(producer1);
            service.execute(producer2);
            service.execute(producer3);
            service.execute(consumer1);
            service.execute(consumer2);
            service.execute(consumer3);
    
            // 3. 睡一会儿，然后尝试停止生产者（结束循环）
            Thread.sleep(10 * 1000);
            producer1.stop();
            producer2.stop();
            producer3.stop();
    
            // 4. 再睡一会儿，关闭线程池
            Thread.sleep(3000);
    
            // 5. shutdown() 等待任务执行完才中断线程(消费者线程未停止，所以其实程序是无法结束的)
            service.shutdown();
        }
    }

因为队列中添加和删除的操作比较频繁，所以这里使用**`LinkedBlockingQueue`**来作为阻塞队列，所以这里除了内存缓冲区有所不同以外，其他的都差不多...当然你也可以指定一个队列的大小。

--------------------

## 三、总结以及改进 ##

生产者-消费者模式很好地对生产者线程和消费者线程进行解耦，优化了系统整体的结构，同时由于缓冲区的作用，允许生产者线程和消费者线程存在执行上的性能差异，从一定程度上缓解了性能瓶颈对系统性能的影响。

上面两种写法都是非常常规的写法，只能说能起码能在及格的基础上加个那么点儿分数，如果想要得高分可以去搜索搜搜 Disruptor 来实现一个无锁的生产者-消费者模型....这里就不提及了..

**改进：**上面的线程输出可能会有点儿不友好（不直观），因为我们这里是直接使用的线程的 ID 来作为输出，我们也可以给线程弄一个名字来作为输出。

[https_www.jianshu.com_p_3f0cd7af370d]: https://www.jianshu.com/p/3f0cd7af370d
[Image 1]: 
[Link 1]: https://zh.wikipedia.org/wiki/%E5%A4%9A%E7%BA%BF%E7%A8%8B
[Link 2]: https://zh.wikipedia.org/wiki/%E5%90%8C%E6%AD%A5
[Link 3]: https://zh.wikipedia.org/wiki/%E7%BC%93%E5%86%B2%E5%8C%BA
[879896-20160602152903211-273313581.png]: /images/20220216/f19e25f9ce224c54a0dfa7dea303697a.png