Java多线程之生产者消费者模式详解

/**
 * 消息队列中存放的消息类
 */
final public class Message {
    private int id;
    private int value;
    public Message(int id,int value){
        this.id = id;
        this.value = value;
    }
    public int getId() {
        return id;
    }
    public int getValue() {
        return value;
    }
}

import lombok.extern.slf4j.Slf4j;
import java.util.LinkedList;
/**
 * 实现一个阻塞队列（消息队列），实现java线程间通信
 */
@Slf4j
public class MessageQueue {
    // 消息队列的容量
    private int capacity;
    // 消息队列
    LinkedList<Message> messageQueue = new LinkedList<>();
    // 设置消息队列的容量
    public MessageQueue(int capacity){
        this.capacity = capacity;
    }
    // 从消息队列中取消息
    public Message take(){
        synchronized (messageQueue){
            // 如果消息队列为空
            while (messageQueue.isEmpty()){
                try {
                    log.debug("队列为空, 消费者线程等待");
                    messageQueue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Message message = messageQueue.removeFirst();
            log.debug("已消费消息 {}", message);
            // 走到这，说明消息队列不为null
            messageQueue.notifyAll();
            return message;
        }
    }
    // 往消息队列中放消息
    public void put(Message message){
        synchronized (messageQueue){
            // 如果消息队列已满
            while (messageQueue.size()==capacity){
                try {
                    log.debug("队列已满, 生产者线程等待");
                    messageQueue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            messageQueue.addLast(message);
            log.debug("已生产消息 {}", message);
            // 走到这，说明消息队列不满
            messageQueue.notifyAll();
        }
    }
}

public class Main {
    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));
            },"生产者").start();
        }
        new Thread(()->{
            while (true){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Message message = queue.take();
            }
        },"消费者").start();
    }
}