image.png

要点

  • 与保护性暂停中的 GuardObject 不同，不需要产生结果和消费结果的线程一一对应
  • 消费队列可以用来平衡生产和消费的线程资源
  • 生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
  • 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
  • JDK 中各种阻塞队列，采用的就是这种模式
  1. public class TestProducerConsumer {
  2.     public static void main(String[] args) {
  3.         MessageQueue messageQueue = new MessageQueue(2);
  4.         for (int i = 0; i < 4; i++) {
  5.             int id = i;
  6.             new Thread(() -> {
  7.                 try {
  8.                     log.debug("download...");
  9.                     List<String> response = Downloader.download();
  10.                     log.debug("try put message({})", id);
  11.                     messageQueue.put(new Message(id, response));
  12.                 } catch (IOException e) {
  13.                     e.printStackTrace();
  14.                 }
  15.             }, "生产者" + i).start();
  16.         }
  18.         new Thread(() -> {
  19.             while (true) {
  20.                 Message message = messageQueue.take();
  21.                 List<String> response = (List<String>) message.getMessage();
  22.                 log.debug("take message({}): [{}] lines", message.getId(), response.size());
  23.             }
  25.         }, "消费者").start();
  26.     }
  27. }
  29. class Message {
  30.     private int id;
  31.     private Object message;
  33.     public Message(int id, Object message) {
  34.         this.id = id;
  35.         this.message = message;
  36.     }
  38.     public int getId() {
  39.         return id;
  40.     }
  42.     public Object getMessage() {
  43.         return message;
  44.     }
  45. }
  47. @Slf4j(topic = "c.MessageQueue")
  48. class MessageQueue {
  49.     private LinkedList<Message> queue;
  50.     private int capacity;
  52.     public MessageQueue(int capacity) {
  53.         this.capacity = capacity;
  54.         queue = new LinkedList<>();
  55.     }
  57.     public Message take() {
  58.         synchronized (queue) {
  59.             while (queue.isEmpty()) {
  60.                 log.debug("没货了, wait");
  61.                 try {
  62.                     queue.wait();
  63.                 } catch (InterruptedException e) {
  64.                     e.printStackTrace();
  65.                 }
  66.             }
  67.             Message message = queue.removeFirst();
  68.             queue.notifyAll();
  69.             return message;
  70.         }
  71.     }
  73.     public void put(Message message) {
  74.         synchronized (queue) {
  75.             while (queue.size() == capacity) {
  76.                 log.debug("库存已达上限, wait");
  77.                 try {
  78.                     queue.wait();
  79.                 } catch (InterruptedException e) {
  80.                     e.printStackTrace();
  81.                 }
  82.             }
  83.             queue.addLast(message);
  84.             queue.notifyAll();
  85.         }
  86.     }
  87. }