Java多線程之生產(chǎn)者消費者模式詳解

更新時間：2022年03月01日 16:15:15 作者：小小茶花女

這篇文章主要為大家詳細(xì)介紹了Java多線程之生產(chǎn)者消費者模式，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助

1. 生產(chǎn)者消費者模型

在生產(chǎn)者-消費者模式中，通常有兩類線程，即生產(chǎn)者線程（若干個）和消費者線程（若干個）。生產(chǎn)者線程向消息隊列加入數(shù)據(jù)，消費者線程則從消息隊列消耗數(shù)據(jù)。生產(chǎn)者和消費者、消息隊列之間的關(guān)系結(jié)構(gòu)圖如圖：

在這里插入圖片描述

(1) 消息隊列可以用來平衡生產(chǎn)和消費的線程資源；

(2) 生產(chǎn)者僅負(fù)責(zé)產(chǎn)生結(jié)果數(shù)據(jù)，不關(guān)心數(shù)據(jù)該如何處理，而消費者專心處理結(jié)果數(shù)據(jù) ；

(3) 消息隊列是有容量限制的，消息隊列滿后，生產(chǎn)者不能再加入數(shù)據(jù)；消息隊列空時，消費者不能再取出數(shù)據(jù)；

(4) 消息隊列是線程安全的，在并發(fā)操作消息隊列的過程中，不能出現(xiàn)數(shù)據(jù)不一致的情況；或者在多個線程并發(fā)更改共享數(shù)據(jù)后，不會造成出現(xiàn)臟數(shù)據(jù)的情況；

(5) JDK 中各種阻塞隊列，采用的就是這種模式；

2. 實現(xiàn)生產(chǎn)者消費者模型

1、消息隊列中存放的消息類：

/**
 * 消息隊列中存放的消息類
 */
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;
    }
}

2、實現(xiàn)阻塞隊列（消息隊列）：

import lombok.extern.slf4j.Slf4j;
import java.util.LinkedList;
/**
 * 實現(xiàn)一個阻塞隊列（消息隊列），實現(xiàn)java線程間通信
 */
@Slf4j
public class MessageQueue {
    // 消息隊列的容量
    private int capacity;
    // 消息隊列
    LinkedList<Message> messageQueue = new LinkedList<>();
    // 設(shè)置消息隊列的容量
    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("隊列已滿, 生產(chǎn)者線程等待");
                    messageQueue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            messageQueue.addLast(message);
            log.debug("已生產(chǎn)消息 {}", message);
            // 走到這，說明消息隊列不滿
            messageQueue.notifyAll();
        }
    }
}

3、測試：

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));
            },"生產(chǎn)者").start();
        }
        new Thread(()->{
            while (true){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Message message = queue.take();
            }
        },"消費者").start();
    }
}

執(zhí)行結(jié)果：

15:31:28.488 [生產(chǎn)者] DEBUG com.example.test.MessageQueue - 已生產(chǎn)消息 com.example.test.Message@54309a75
15:31:28.507 [生產(chǎn)者] DEBUG com.example.test.MessageQueue - 已生產(chǎn)消息 com.example.test.Message@50915389
15:31:28.507 [生產(chǎn)者] DEBUG com.example.test.MessageQueue - 隊列已滿, 生產(chǎn)者線程等待
15:31:29.486 [消費者] DEBUG com.example.test.MessageQueue - 已消費消息 com.example.test.Message@54309a75
15:31:29.486 [生產(chǎn)者] DEBUG com.example.test.MessageQueue - 已生產(chǎn)消息 com.example.test.Message@6340ac12
15:31:30.487 [消費者] DEBUG com.example.test.MessageQueue - 已消費消息 com.example.test.Message@50915389
15:31:31.487 [消費者] DEBUG com.example.test.MessageQueue - 已消費消息 com.example.test.Message@6340ac12
15:31:32.488 [消費者] DEBUG com.example.test.MessageQueue - 隊列為空, 消費者線程等待