设计模式 Concurrency 之 ProducerConsumer 生成者消费者模式

小鱼儿 2022-06-10 03:25 289阅读 0赞

*  定义
 *  优缺点
 *  例子

# 1. 定义 #

某个模块负责产生数据，这些数据由另一个模块来负责处理（此处的模块是广义的，可以是类、函数、线程、进程等）。产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者。

# 2. 优缺点 #

优点:

*  解耦  
    
    
     *  假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化，可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。
 *  支持并发  
    
    
     *  生产者和消费者可以是两个独立的并发主体,生产者只需要把数据放入数据缓冲区便可继续去生产,不用依赖于消费的处理速度

# 3. 例子 #

![这里写图片描述][SouthEast]

Item

package com.hqq.concurrency.producer_consumer;
    
    /** * Item * Created by heqianqian on 2017/8/1. */
    public class Item { 
    
        private String producer;
    
        private Integer id;
    
        public Item(String producer, Integer id) {
            this.producer = producer;
            this.id = id;
        }
    
        public String getProducer() {
            return producer;
        }
    
        public Integer getId() {
            return id;
        }
    }

ItemQueue\[内部使用的是BlockingQueue阻塞队列 当队列中没有元素 会阻塞等待\]

package com.hqq.concurrency.producer_consumer;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    /** * ItemQueue * Created by heqianqian on 2017/8/1. */
    public class ItemQueue { 
    
        private BlockingQueue<Item> queue;
    
        public ItemQueue() {
            queue = new LinkedBlockingQueue<>(5);
        }
    
        public void put(Item item) throws InterruptedException {
            queue.put(item);
        }
    
        public Item take() throws InterruptedException {
            return queue.take();
        }
    
    }

Producer

package com.hqq.concurrency.producer_consumer;
    
    import java.util.Random;
    
    /** * Producer * 生产者 * Created by heqianqian on 2017/8/1. */
    public class Producer { 
    
        private final ItemQueue queue;
    
        private final String name;
    
        private int itemId;
    
        public Producer(String name, ItemQueue queue) {
            this.name = name;
            this.queue = queue;
        }
    
        /** * Put item in the queue */
        public void produce() throws InterruptedException {
    
            Item item = new Item(name, itemId++);
            queue.put(item);
            Random random = new Random();
            Thread.sleep(random.nextInt(2000));
        }
    }

Consumer

package com.hqq.concurrency.producer_consumer;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * Consumer * 消费者 * Created by heqianqian on 2017/8/1. */
    public class Consumer { 
        private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
    
        private final ItemQueue queue;
    
        private final String name;
    
        public Consumer(String name, ItemQueue queue) {
            this.name = name;
            this.queue = queue;
        }
    
        /** * Consume item from the queue */
        public void consume() throws InterruptedException {
    
            Item item = queue.take();
            LOGGER.info("Consumer [{}] consume item [{}] produced by [{}]", name, item.getId(), item.getProducer());
    
        }
    }

APP

package com.hqq.concurrency.producer_consumer;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /** * App * Created by heqianqian on 2017/8/1. */
    public class App { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(App.class);
    
        public static void main(String... args) {
    
            ItemQueue queue = new ItemQueue();
    
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 2; i++) {
    
                final Producer producer = new Producer("Producer_" + i, queue);
                executorService.submit(() -> {
                    while (true) {
                        producer.produce();
                    }
                });
            }
    
            for (int i = 0; i < 3; i++) {
                final Consumer consumer = new Consumer("Consumer_" + i, queue);
                executorService.submit(() -> {
                    while (true) {
                        consumer.consume();
                    }
                });
            }
    
            executorService.shutdown();
            try {
                executorService.awaitTermination(10, TimeUnit.SECONDS);
                executorService.shutdownNow();
            } catch (InterruptedException e) {
                LOGGER.error("Error waiting for ExecutorService shutdown");
            }
        }
    
    }

运行结果：

INFO  [2017-08-09 02:40:51,891] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_0] consume item [0] produced by [Producer_1]
    INFO  [2017-08-09 02:40:51,891] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_1] consume item [0] produced by [Producer_0]
    INFO  [2017-08-09 02:40:52,270] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_2] consume item [1] produced by [Producer_1]
    INFO  [2017-08-09 02:40:53,085] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_0] consume item [1] produced by [Producer_0]
    INFO  [2017-08-09 02:40:53,822] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_1] consume item [2] produced by [Producer_0]
    INFO  [2017-08-09 02:40:53,853] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_2] consume item [2] produced by [Producer_1]
    INFO  [2017-08-09 02:40:54,721] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_0] consume item [3] produced by [Producer_1]
    INFO  [2017-08-09 02:40:55,474] com.hqq.concurrency.producer_consumer.Consumer: Consumer [Consumer_1] consume item [4] produced by [Producer_1]
    ...

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

[SouthEast]: /images/20220610/de31632f07c94813b2b4131729adfa7f.png