一文吃透生產者和消費者模型！

作者：志哥聊技術 2024-10-11 09:27:52

開發前端

簡單的說，生產者和消費者之間不直接進行交互，而是通過一個緩沖區來進行交互，生產者負責生成數據，然后存入緩沖區；消費者則負責處理數據，從緩沖區獲取。

01、背景介紹

在 Java 多線程編程中，還有一個非常重要的設計模式，它就是：生產者和消費者模型。

這種模型可以充分發揮 cpu 的多線程特性，通過一些平衡手段能有效的提升系統整體處理數據的速度，減輕系統負載，提高程序的效率和穩定性，同時實現模塊之間的解耦。

那什么是生產者和消費者模型呢？

大致流程圖如下：

圖片

對于最簡單的生產者和消費者模型，總結下來，大概有以下幾個特點：

緩沖區為空的時候，消費者不能消費，會進入休眠狀態，直到有新數據進入緩沖區，再次被喚醒
緩沖區填滿的時候，生產者不能生產，也會進入休眠狀態，直到緩沖區有空間，再次被喚醒

生產者和消費者模型作為一個非常重要的設計模型，它的優點在于：

解耦：生產者和消費者之間不直接進行交互，即使生產者和消費者的代碼發生變化，也不會對對方產生影響
消峰：例如在某項工作中，假如 A 操作生產數據的速度很快，B 操作處理速度很慢，那么 A 操作就必須等待 B 操作完成才能結束，反之亦然。如果將 A 操作和B 操作進行解耦，中間插入一個緩沖區，這樣 A 操作將生產的數據存入緩沖區，就接受了；B 操作從緩沖區獲取數據并進行處理，平衡好 A 操作和 B 操作之間的緩沖區，可以顯著提升系統的數據處理能力

生產者和消費者模型的應用場景非常多，例如 Java 的線程池任務執行框架、消息中間件 rabbitMQ 等，因此掌握生產者和消費者模型，對于開發者至關重要。

下面我們通過幾個案例，一起來了解一下生產者和消費者設計模型的實踐思路。

02、代碼實踐

2.1、利用 wait / notify 方法實現思路

生產者和消費者模型，最簡單的一種技術實踐方案就是基于線程的 wait() / notify() 方法，也就是通知和喚醒機制，可以將兩個操作實現解耦，具體代碼實踐如下。

/**
 * 緩沖區容器類
 */
public class Container {

    /**
     * 緩沖區最大容量
     */
    private int capacity = 3;

    /**
     * 緩沖區
     */
    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩沖區
     * @param value
     */
    public synchronized void add(Integer value) {
        if(list.size() >= capacity){
            System.out.println("生產者："+ Thread.currentThread().getName()+"，緩沖區已滿,生產者進入waiting...");
            try {
                // 進入等待狀態
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
        list.add(value);

        //喚醒其他所有處于wait()的線程，包括消費者和生產者
        notifyAll();
    }


    /**
     * 從緩沖區獲取數據
     */
    public synchronized void get() {
        if(list.size() == 0){
            System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區為空,消費者進入waiting...");
            try {
                // 進入等待狀態
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 從頭部獲取數據，并移除元素
        Integer val = list.removeFirst();
        System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

        //喚醒其他所有處于wait()的線程，包括消費者和生產者
        notifyAll();
    }
}

/**
 * 生產者類
 */
public class Producer extends Thread{

    private Container container;

    public Producer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.add(i);
        }
    }
}

/**
 * 消費者類
 */
public class Consumer extends Thread{

    private Container container;

    public Consumer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.get();
        }
    }
}

/**
 * 測試類
 */
public class MyThreadTest {

    public static void main(String[] args) {
        Container container = new Container();
        Producer producer = new Producer(container);
        Consumer consumer = new Consumer(container);

        producer.start();
        consumer.start();
    }
}

運行結果如下：

生產者：Thread-0，add：0
生產者：Thread-0，add：1
生產者：Thread-0，add：2
生產者：Thread-0，緩沖區已滿,生產者進入waiting...
消費者：Thread-1，value：0
消費者：Thread-1，value：1
消費者：Thread-1，value：2
消費者：Thread-1，緩沖區為空,消費者進入waiting...
生產者：Thread-0，add：3
生產者：Thread-0，add：4
生產者：Thread-0，add：5
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5

從日志上可以很清晰的看到，生產者線程生產一批數據之后，當緩沖區已經滿了，會進入等待狀態，此時會通知消費者線程；消費者線程處理完數據之后，當緩沖區沒有數據時，也會進入等待狀態，再次通知生產者線程。

2.2、利用 await / signal 方法實現思路

除此之外，我們還可以利用ReentrantLock和Condition類中的 await() / signal() 方法實現生產者和消費者模型。

緩沖區容器類，具體代碼實踐如下。

/**
 * 緩沖區容器類
 */
public class Container {

    private Lock lock = new ReentrantLock();

    private Condition condition = lock.newCondition();

    private int capacity = 3;

    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩沖區
     * @param value
     */
    public void add(Integer value) {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() >= capacity){
                System.out.println("生產者："+ Thread.currentThread().getName()+"，緩沖區已滿,生產者進入waiting...");
                // 進入等待狀態
                condition.await();
            }
            System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
            list.add(value);

            //喚醒其他所有處于wait()的線程，包括消費者和生產者
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }


    /**
     * 從緩沖區獲取數據
     */
    public void get() {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() == 0){
                System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區為空,消費者進入waiting...");
                // 進入等待狀態
                condition.await();
            }
            // 從頭部獲取數據，并移除元素
            Integer val = list.removeFirst();
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

            //喚醒其他所有處于wait()的線程，包括消費者和生產者
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }
}

生產者、消費者、測試類代碼，跟上文一致，運行結果和上文介紹的也是一樣。

2.3、多生產者和消費者的實現思路

上面介紹的都是一個生產者線程和一個消費者線程，模型比較簡單。實際上，在業務開發中，經常會出現多個生產者線程和多個消費者線程，按照以上的實現思路，會出現什么情況呢？

有可能會出現程序假死現象！下面我們來分析一下案例，假如有兩個生產者線程 a1、a2，兩個消費者線程 b1、b2，執行過程如下：

1.生產者線程 a1 執行生產數據的操作，發現緩沖區數據已經填滿了，然后進入等待階段，同時向外發起通知，喚醒其它線程
2.因為線程喚醒具有隨機性，本應該喚醒消費者線程 b1，結果可能生產者線程 a2 被喚醒，檢查緩沖區數據已經填滿了，又進入等待階段，緊接向外發起通知，消費者線程得不到被執行的機會
3.消費者線程 b1、b2，也有可能會出現這個現象，本應該喚醒生產者線程，結果喚醒了消費者線程

遇到這種情況，應該如何解決呢？

因為ReentrantLock和Condition的結合，編程具有高度靈活性，我們可以采用這種組合解決多生產者和多消費者中的假死問題。

具體實現邏輯如下：

/**
 * 緩沖區容器類
 */
public class ContainerDemo {

    private Lock lock = new ReentrantLock();
    private Condition producerCondition = lock.newCondition();
    private Condition consumerCondition = lock.newCondition();

    private int capacity = 3;
    private LinkedList<Integer> list = new LinkedList<Integer>();


    /**
     * 添加數據到緩沖區
     * @param value
     */
    public void add(Integer value) {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() >= capacity){
                System.out.println("生產者："+ Thread.currentThread().getName()+"，緩沖區已滿,生產者進入waiting...");
                // 生產者進入等待狀態
                producerCondition.await();
            }
            System.out.println("生產者："+ Thread.currentThread().getName()+"，add：" + value);
            list.add(value);

            // 喚醒所有消費者處于wait()的線程
            consumerCondition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }


    /**
     * 從緩沖區獲取數據
     */
    public void get() {
        boolean flag = false;
        try {
            flag = lock.tryLock(3, TimeUnit.SECONDS);
            if(list.size() == 0){
                System.out.println("消費者："+ Thread.currentThread().getName()+"，緩沖區為空,消費者進入waiting...");
                // 消費者進入等待狀態
                consumerCondition.await();
            }
            // 從頭部獲取數據，并移除元素
            Integer val = list.removeFirst();
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + val);

            // 喚醒所有生產者處于wait()的線程
            producerCondition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(flag){
                lock.unlock();
            }
        }
    }
}

/**
 * 生產者
 */
public class Producer extends Thread{

    private ContainerDemo container;

    private Integer value;

    public Producer(ContainerDemo container, Integer value) {
        this.container = container;
        this.value = value;
    }

    @Override
    public void run() {
        container.add(value);
    }
}

/**
 * 消費者
 */
public class Consumer extends Thread{

    private ContainerDemo container;

    public Consumer(ContainerDemo container) {
        this.container = container;
    }

    @Override
    public void run() {
        container.get();
    }
}

/**
 * 測試類
 */
public class MyThreadTest {

    public static void main(String[] args) {
        ContainerDemo container = new ContainerDemo();

        List<Thread> threadList = new ArrayList<>();
        // 初始化6個生產者線程
        for (int i = 0; i < 6; i++) {
            threadList.add(new Producer(container, i));
        }
        // 初始化6個消費者線程
        for (int i = 0; i < 6; i++) {
            threadList.add(new Consumer(container));
        }

        // 啟動線程
        for (Thread thread : threadList) {
            thread.start();
        }
    }
}

運行結果如下：

生產者：Thread-0，add：0
生產者：Thread-1，add：1
生產者：Thread-2，add：2
生產者：Thread-3，緩沖區已滿,生產者進入waiting...
生產者：Thread-4，緩沖區已滿,生產者進入waiting...
生產者：Thread-5，緩沖區已滿,生產者進入waiting...
消費者：Thread-6，value：0
消費者：Thread-7，value：1
生產者：Thread-3，add：3
生產者：Thread-4，add：4
生產者：Thread-5，add：5
消費者：Thread-8，value：2
消費者：Thread-9，value：3
消費者：Thread-10，value：4
消費者：Thread-11，value：5

通過ReentrantLock定義兩個Condition，一個表示生產者的Condition，一個表示消費者的Condition，喚醒的時候調用對應的signalAll()方法就可以解決假死現象。

03、小結

最后我們來總結一下，對于生產者和消費者模型，通過合理的編程實現，可以充分充分發揮 cpu 多線程的特性，顯著的提升系統處理數據的效率。

對于生產者和消費者模型中的假死現象，可以使用ReentrantLock定義兩個Condition，進行交叉喚醒，以解決假死問題。

責任編輯：武曉燕來源：潘志的研發筆記

生產者消費者模型

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看