异步模式：生产者&消费者

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

消息队列

消息类：生产者和消费者通信的对象


@Getter  
final class Message {  
    private final int id;  
    private final Object value;  
  
    public Message(int id, Object value) {  
        this.id = id;  
        this.value = value;  
    }  
      
    @Override  
    public String toString() {  
        return "Message{" +  
                "id=" + id +  
                ", value=" + value +  
                '}';  
    }  
}

消息队列：生产者和消费者通信的操作


class MessageQueue {  
    // 消息的队列集合  
    private final LinkedList<Message> list = new LinkedList<>();  
    // 队列容量  
    private final int capcity;  
  
    public MessageQueue(int capcity) {  
        this.capcity = capcity;  
    }  
  
    // 获取消息  
    public Message take() {  
        // 检查队列是否为空  
        synchronized (list) {  
            while (list.isEmpty()) {  
                try {  
                    log.debug("队列为空, 消费者线程等待");  
                    list.wait();  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
            // 从队列头部获取消息并返回  
            Message message = list.removeFirst();  
            log.debug("已消费消息 {}", message);  
            list.notifyAll();  
            return message;  
        }  
    }  
  
    // 存入消息  
    public void put(Message message) {  
        synchronized (list) {  
            // 检查对象是否已满  
            while (list.size() == capcity) {  
                try {  
                    log.debug("队列已满, 生产者线程等待");  
                    list.wait();  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
            // 将消息加入队列尾部  
            list.addLast(message);  
            log.debug("已生产消息 {}", message);  
            // 通知消费者有消息可以消费了
            list.notifyAll();  
        }  
    }  
}

测试生产者和消费者：


public static void main(String[] args) {  
    MessageQueue queue = new MessageQueue(2);  
    for (int i = 0; i < 3; i++) {  
        int id = i;  
        new Thread(() -> {  
            queue.put(new Message(id, "值" + id));  
        }, "生产者" + i).start();  
    }  
    new Thread(() -> {  
        while (true) {  
            sleep(1);  
            Message message = queue.take();  
        }  
    }, "消费者").start();  
}