模式简介

生产者&消费者模式是线程进进行通信的一种异步通信方式，之所以说它是异步的，是因为生产者生成的数据可以存放到上图中的一个固定大小缓冲区中，而这个缓冲区一般采用“先进先出”的队列来实现，因此生成者put数据之后，消费者不一定可以立刻拿到数据。而上一篇的保护性暂停模式则是一种同步的通信方式，并且只能传递一个对象，而这里的模式可以传递多个对象。我们可以来看一下这个模式具有的一些优点：

1. 解耦：该模式实现了消费者和生产者之间的解耦，生产者只需要生成数据，不需要像保护性暂停模式那样还要管给谁生成数据；同样，消费者也只需要管消费数据就行了，不需要管这个数据来源与哪里

1. 异步：而异步就是上面所结束的一样，无需保证数据一旦产生就必须消费的同步模式

1. 平衡生成和消费的速度差：如果消费者消费的很快，就可以多放几个生产者生成数据；同样，如果消费者消费的很慢，少放几个生产者就可以了

实现代码

@Slf4j(topic = "c.MessageQueue")
public class MessageQueue {
    //消息队列，缓冲区
    private final LinkedList<Message> queue;
    //缓冲区容量
    private final 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();
        }
    }
    public static void main(String[] args) {
        MessageQueue messageQueue = new MessageQueue(2);
        // 4 个生产者线程, 下载任务,返回下载资源response
        for (int i = 0; i < 4; i++) {
            int id = i;
            new Thread(() -> {
                try {
                    log.debug("download...");
                    //模拟下载资源
                    int time = id * 1000;
                    Thread.sleep(time);
                    String response = "下载的资源，资源标识：" + id;
                    log.debug("try put message {}", response);
                    messageQueue.put(new Message(id, response));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "生产者" + i).start();
        }
        // 1 个消费者线程, 处理结果
        new Thread(() -> {
            while (true) {
                Message message = messageQueue.take();
                String response = (String) message.getMessage();
                log.debug("take message( [{}]->[{}] )", message.getId(), response);
            }
        }, "消费者").start();
    }
}
final class Message {
    private final int id;
    private final Object message;
    public Message(int id, Object message) {
        this.id = id;
        this.message = message;
    }
    public int getId() {
        return id;
    }
    public Object getMessage() {
        return message;
    }
}

执行结果：