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