异步模式之生产者消费者

4.异步模式之生产者消费者

刚才的情况，我们一直都在讨论，一个执行者对应一个接收者的情况。下面，我们来讨论另外一种情况，多个生产者对应多应多个消费者的情况。

//消息类
class Message {
 
    private int id;
 
    private Object message;
 
 public Message(int id, Object message) {
 
     this.id = id;
 
     this.message = message;
 }
 
 public int getId() {
 
     return id;
 }
 
public Object getMessage() {

    return message;
 }

}
//消息队列
class MessageQueue {
//为了方便，我们采用双端队列
    private LinkedList<Message> queue;

    private int capacity;
 
 public MessageQueue(int capacity) {
 
     this.capacity = capacity;
 
     queue = new LinkedList<>();
 }
 //获取消息
 public Message take() {
 
 synchronized (queue) {
 
   while (queue.isEmpty()) {
 
       log.debug("没货了, wait");
 try {
 
     queue.wait();
 } catch (InterruptedException e) {
 
     e.printStackTrace();
 }
 }
 //我们获取完了要提醒提供者
     Message message = queue.removeFirst();

     queue.notifyAll();
 
     return message;
 }
 }
 //存放消息
 public void put(Message message) {
 
  synchronized (queue) {
 
    while (queue.size() == capacity) {
 
        log.debug("库存已达上限, wait");
 try {
 
     queue.wait();
 } catch (InterruptedException e) {

     e.printStackTrace();
 }
 }
 //提供者提供完了也要提醒消费者
      queue.addLast(message);

      queue.notifyAll();
 }
 }
}

• 测试

MessageQueue messageQueue = new MessageQueue(2);
// 4 个生产者线程, 下载任务
for (int i = 0; i < 4; i++) {
 
    int id = i;
 
 new Thread(() -> {
 try {
 
     log.debug("download...");
 
     List<String> response = Downloader.download();
 
     log.debug("try put message({})", id);
 
     messageQueue.put(new Message(id, response));
 } catch (IOException e) {
 
     e.printStackTrace();
 }
 }, "生产者" + i).start();
}

// 1 个消费者线程, 处理结果

new Thread(() -> {
 while (true) {

     Message message = messageQueue.take();
 
     List<String> response = (List<String>) message.getMessage();
 
     log.debug("take message({}): [{}] lines", message.getId(), response.size());
 }
}, "消费者").start();