Java并发基础九：并发协作(生产者消费者模型)

淡淡的烟草味﹌ 2023-07-24 13:18 57阅读 0赞

**对于多线程程序来说，不管任何编程语言，生产者和消费者模型都是最经典的。就像学习每一门编程语言一样，Hello World！都是最经典的例子。**  
  
实际上，准确说应该是“生产者-消费者-仓储”模型，离开了仓储，生产者消费者模型就显得没有说服力了。  
对于此模型，应该明确一下几点：  
**1、生产者仅仅在仓储未满时候生产，仓满则停止生产。  
2、消费者仅仅在仓储有产品时候才能消费，仓空则等待。  
3、当消费者发现仓储没产品可消费时候会通知生产者生产。  
4、生产者在生产出可消费产品时候，应该通知等待的消费者去消费。**

**Demo实列：**

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * @author lishujiang
     * @date 2020/4/11 15:11
     */
    public class ProducerConsumer {
        //代表生产的商品数量
        private static AtomicInteger count = new AtomicInteger(0);
    
        /**
         * 消费者
         */
        class Consumer implements Runnable {
            private String name;
            private Storage s = null;
    
            public Consumer(String name, Storage s) {
                this.name = name;
                this.s = s;
            }
    
            public void run() {
                try {
                    while (true) {
                        Thread.sleep(5001);
                        System.out.println(name + "准备消费产品.");
                        Product product = s.pop();
                        System.out.println(name + "已消费(" + product.toString() + ").");
                        System.out.println("===============");
                        Thread.sleep(500);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        /**
         * 生产者
         */
        class Producer implements Runnable {
            private String name;
            private Storage s = null;
    
            public Producer(String name, Storage s) {
                this.name = name;
                this.s = s;
            }
    
            public void run() {
                try {
                    while (true) {
                        Product product = new Product(count.incrementAndGet()); // 产生产品
                        System.out.println(name + "准备生产(" + product.toString() + ").");
                        s.push(product);
                        System.out.println(name + "已生产(" + product.toString() + ").");
                        System.out.println("===============");
                        Thread.sleep(500);
                    }
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
    
            }
        }
    
        /**
         * 仓库容量默认设置为10，阻塞队列生产消费线程安全
         */
        public class Storage {
            BlockingQueue<Product> queues = new LinkedBlockingQueue<Product>(10);
    
            /**
             * 生产
             *
             * @param p 产品
             * @throws InterruptedException
             */
            public void push(Product p) throws InterruptedException {
                queues.put(p);
            }
    
            /**
             * 消费
             *
             * @return 产品
             * @throws InterruptedException
             */
            public Product pop() throws InterruptedException {
                return queues.take();
            }
        }
    
        /**
         * 产品
         */
        public class Product {
            private int id;
    
            public Product(int id) {
                this.id = id;
            }
    
            public String toString() {// 重写toString方法
                return "产品：" + this.id;
            }
        }
    
        public static void main(String[] args) {
            ProducerConsumer pc = new ProducerConsumer();
    
            Storage s = pc.new Storage();
    
            ExecutorService service = Executors.newCachedThreadPool();
            Producer p = pc.new Producer("张三", s);
            Producer p2 = pc.new Producer("李四", s);
            Consumer c = pc.new Consumer("王五", s);
            Consumer c2 = pc.new Consumer("老刘", s);
            Consumer c3 = pc.new Consumer("老林", s);
            service.submit(p);
            service.submit(p2);
            service.submit(c);
    //        service.submit(c2);
    //        service.submit(c3);
        }
    }

运行结果

张三准备生产(产品：1).
    张三已生产(产品：1).
    ===============
    李四准备生产(产品：2).
    李四已生产(产品：2).
    ===============
    张三准备生产(产品：3).
    张三已生产(产品：3).
    ===============
    李四准备生产(产品：4).
    李四已生产(产品：4).
    ===============
    张三准备生产(产品：5).
    张三已生产(产品：5).
    ===============
    李四准备生产(产品：6).
    李四已生产(产品：6).
    ===============
    李四准备生产(产品：7).
    李四已生产(产品：7).
    ===============
    张三准备生产(产品：8).
    张三已生产(产品：8).
    ===============
    李四准备生产(产品：9).
    李四已生产(产品：9).
    ===============
    张三准备生产(产品：10).
    张三已生产(产品：10).
    ===============
    李四准备生产(产品：11).
    张三准备生产(产品：12).
    王五准备消费产品.
    李四已生产(产品：11).
    ===============
    王五已消费(产品：1).
    ===============
    李四准备生产(产品：13).

## [阻塞队列BlockingQueue用法][BlockingQueue] ##

多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。

假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。

然而，在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒）

BlockingQueue的两个常见阻塞场景：  
1、当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。  
2、当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空闲的位置，线程被自动唤醒。

这也是我们在多线程环境下，为什么需要BlockingQueue的原因。作为BlockingQueue的使用者，我们再也不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切BlockingQueue都给你一手包办了。

阻塞队列参考：[https://www.cnblogs.com/linjiqin/p/5130559.html][BlockingQueue]

[BlockingQueue]: https://www.cnblogs.com/linjiqin/p/5130559.html