外链数是网站反向链接码/个人博客

1. 定义

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

 2. 实现

//消息
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();   //共享资源不足，对其上锁之后进入其的休息室等待（waitSet）} catch (InterruptedException e) { //此时消费者线程处于TIMED_WAITING状态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();   //唤醒未获取到资源而处于休息室等待的消费者线程}}
}

3. 测试

 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();

结果：

        4个生产者下载任务，消费者线程未获取到资源，wait。生产者线程都下载好后，尝试存入消息队列，由于容量为2，只存入两个消息，生产者2和1线程wait。之后消费者逐渐消费完所有资源。