1. 定义
    要点

    • 与前面的保护性暂停中的 GuardObject 不同，不需要产生结果和消费结果的线程一一对应
    • 消费队列可以用来平衡生产和消费的线程资源
    • 生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
    • 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
    • JDK 中各种阻塞队列，采用的就是这种模式
    • 阻塞队列有：BlockingQueue

    异步模式之生产者/消费者(Queue原理) - 图1
    BlockingQueue的源码

    1. // 队列项
    2. final Object[] items;
    3. // 保护所有通道的主锁
    4. final ReentrantLock lock;
    5. // 等待条件
    6. private final Condition notEmpty;
    7. // 等待插入锁
    8. private final Condition notFull;
    9. // 容量
    10. int count;
    11. // 下一个 put、offer 或 add 的项目索引
    12. int putIndex;
    14. // 每加入一个的元素，容量加一
    15. private void enqueue(E x) {
    16.     final Object[] items = this.items;
    17.     items[putIndex] = x;
    18.     if (++putIndex == items.length)
    19.         putIndex = 0;
    20.     count++;
    21.     notEmpty.signal();
    22. }
    24. /***
    25.   * 构造函数
    26.   */
    27. public ArrayBlockingQueue(int capacity, boolean fair) {
    28.     if (capacity <= 0)
    29.         throw new IllegalArgumentException();
    30.     // 将数组的容量赋值
    31.     this.items = new Object[capacity];
    32.     // fair为 true ，先进先出（FIFO）
    33.     lock = new ReentrantLock(fair);
    34.     // 创建两个新的锁
    35.     notEmpty = lock.newCondition();
    36.     notFull =  lock.newCondition();
    37. }
    39. public void put(E e) throws InterruptedException {
    40.     // 检查对象是否为空
    41.     checkNotNull(e);
    42.     // 创建lock锁
    43.     final ReentrantLock lock = this.lock;
    44.     // 等待锁
    45.     lock.lockInterruptibly();
    46.     try {
    47.         // 该队列容量满了
    48.         while (count == items.length)
    49.             // 睡眠
    50.             notFull.await();
    51.         enqueue(e);
    52.     } finally {
    53.         // 唤醒
    54.         lock.unlock();
    55.     }
    56. }
    58.  public E take() throws InterruptedException {
    59.     final ReentrantLock lock = this.lock;
    60.     lock.lockInterruptibly();
    61.     try {
    62.         while (count == 0)
    63.             notEmpty.await();
    64.         return dequeue();
    65.     } finally {
    66.         lock.unlock();
    67.     }
    68. }

    示例

    1. @Slf4j(topic = "c.Test21")
    2. public class Test21 {
    4.     public static void main(String[] args) {
    5.         MessageQueue queue = new MessageQueue(2);
    7.         for (int i = 0; i < 3; i++) {
    8.             int id = i;
    9.             new Thread(() -> {
    10.                 queue.put(new Message(id , "值"+id));
    11.             }, "生产者" + i).start();
    12.         }
    14.         new Thread(() -> {
    15.             while(true) {
    16.                 sleep(1);
    17.                 Message message = queue.take();
    18.             }
    19.         }, "消费者").start();
    20.     }
    22. }
    24. // 消息队列类 ， java 线程之间通信
    25. @Slf4j(topic = "c.MessageQueue")
    26. class MessageQueue {
    27.     // 消息的队列集合
    28.     private LinkedList<Message> list = new LinkedList<>();
    29.     // 队列容量
    30.     private int capcity;
    32.     public MessageQueue(int capcity) {
    33.         this.capcity = capcity;
    34.     }
    36.     // 获取消息
    37.     public Message take() {
    38.         // 检查队列是否为空
    39.         synchronized (list) {
    40.             while(list.isEmpty()) {
    41.                 try {
    42.                     log.debug("队列为空, 消费者线程等待");
    43.                     list.wait();
    44.                 } catch (InterruptedException e) {
    45.                     e.printStackTrace();
    46.                 }
    47.             }
    48.             // 从队列头部获取消息并返回
    49.             Message message = list.removeFirst();
    50.             log.debug("已消费消息 {}", message);
    51.             list.notifyAll();
    52.             return message;
    53.         }
    54.     }
    56.     // 存入消息
    57.     public void put(Message message) {
    58.         synchronized (list) {
    59.             // 检查对象是否已满
    60.             while(list.size() == capcity) {
    61.                 try {
    62.                     log.debug("队列已满, 生产者线程等待");
    63.                     list.wait();
    64.                 } catch (InterruptedException e) {
    65.                     e.printStackTrace();
    66.                 }
    67.             }
    68.             // 将消息加入队列尾部
    69.             list.addLast(message);
    70.             log.debug("已生产消息 {}", message);
    71.             list.notifyAll();
    72.         }
    73.     }
    74. }
    76. final class Message {
    77.     private int id;
    78.     private Object value;
    80.     public Message(int id, Object value) {
    81.         this.id = id;
    82.         this.value = value;
    83.     }
    85.     public int getId() {
    86.         return id;
    87.     }
    89.     public Object getValue() {
    90.         return value;
    91.     }
    93.     @Override
    94.     public String toString() {
    95.         return "Message{" +
    96.                 "id=" + id +
    97.                 ", value=" + value +
    98.                 '}';
    99.     }
    100. }

    结果

    1. 22:58:21.875 c.MessageQueue [生产者1] - 已生产消息 Message{id=1, value=值1}
    2. 22:58:21.878 c.MessageQueue [生产者0] - 已生产消息 Message{id=0, value=值0}
    3. 22:58:21.878 c.MessageQueue [生产者2] - 队列已满, 生产者线程等待
    4. 22:58:22.877 c.MessageQueue [消费者] - 已消费消息 Message{id=1, value=值1}
    5. 22:58:22.877 c.MessageQueue [生产者2] - 已生产消息 Message{id=2, value=值2}
    6. 22:58:23.891 c.MessageQueue [消费者] - 已消费消息 Message{id=0, value=值0}
    7. 22:58:24.906 c.MessageQueue [消费者] - 已消费消息 Message{id=2, value=值2}
    8. 22:58:25.920 c.MessageQueue [消费者] - 队列为空, 消费者线程等待