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

本文主要是介绍设计模式 Concurrency 之 ProducerConsumer 生成者消费者模式，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 定义

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

2. 优缺点

优点:

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

3. 例子

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]
...

---

这篇关于设计模式 Concurrency 之 ProducerConsumer 生成者消费者模式的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---