日韩性视频-久久久蜜桃-www中文字幕-在线中文字幕av-亚洲欧美一区二区三区四区-撸久久-香蕉视频一区-久久无码精品丰满人妻-国产高潮av-激情福利社-日韩av网址大全-国产精品久久999-日本五十路在线-性欧美在线-久久99精品波多结衣一区-男女午夜免费视频-黑人极品ⅴideos精品欧美棵-人人妻人人澡人人爽精品欧美一区-日韩一区在线看-欧美a级在线免费观看

歡迎訪問 生活随笔!

生活随笔

當前位置: 首頁 > 编程资源 > 编程问答 >内容正文

编程问答

多生产者_多线程必考的「生产者 - 消费者」模型,看齐姐这篇文章就够了

發布時間:2025/5/22 编程问答 30 豆豆
生活随笔 收集整理的這篇文章主要介紹了 多生产者_多线程必考的「生产者 - 消费者」模型,看齐姐这篇文章就够了 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

生產者 - 消費者模型 Producer-consumer problem 是一個非常經典的多線程并發協作的模型,在分布式系統里非常常見。也是面試中無論中美大廠都非常愛考的一個問題,對應屆生問的要少一些,但是對于有工作經驗的工程師來說,非常愛考。

這個問題有非常多的版本和解決方式,在本文我重點是和大家壹齊理清思路,由淺入深的思考問題,保證大家看完了都能有所收獲。

問題背景

簡單來說,這個模型是由兩類線程構成:

  • 生產者線程:“生產”產品,并把產品放到一個隊列里;
  • 消費者線程:“消費”產品。

有了這個隊列,生產者就只需要關注生產,而不用管消費者的消費行為,更不用等待消費者線程執行完;消費者也只管消費,不用管生產者是怎么生產的,更不用等著生產者生產。

所以該模型實現了生產者和消費者之間的解藕異步

什么是異步呢?

比如說你和你女朋友打電話,就得等她接了電話你們才能說話,這是同步。

但是如果你跟她發微信,并不需要等她回復,她也不需要立刻回復,而是等她有空了再回,這就是異步。

但是呢,生產者和消費者之間也不能完全沒有聯系的。

  • 如果隊列里的產品已經滿了,生產者就不能繼續生產;
  • 如果隊列里的產品從無到有,生產者就得通知一下消費者,告訴它可以來消費了;
  • 如果隊列里已經沒有產品了,消費者也無法繼續消費;
  • 如果隊列里的產品從滿到不滿,消費者也得去通知下生產者,說你可以來生產了。

所以它們之間還需要有協作,最經典的就是使用 Object 類里自帶的 wait() 和 notify() 或者 notifyAll() 的消息通知機制。

上述描述中的等著,其實就是用 wait() 來實現的;

通知,就是 notify() 或者 notifyAll() 。

那么基于這種消息通知機制,我們還能夠平衡生產者和消費者之間的速度差異

如果生產者的生產速度很慢,但是消費者消費的很快,就像是我們每月工資就發兩次,但是每天都要花錢,也就是 1:15.

那么我們就需要調整生產者(發工資)為 15 個線程,消費者保持 1 個線程,這樣是不是很爽~

總結下該模型的三大優點:
解藕,異步,平衡速度差異。

wait()/notify()

接下來我們需要重點看下這個通知機制。

wait() 和 notify() 都是 Java 中的 Object 類自帶的方法,可以用來實現線程間的通信。

在上一節講的 11 個 APIs 里我也提到了它,我們這里再展開講一下。

wait() 方法是用來讓當前線程等待,直到有別的線程調用 notify() 將它喚醒,或者我們可以設定一個時間讓它自動蘇醒。

調用該方法之前,線程必須要獲得該對象的對象監視器鎖,也就是只能用在加鎖的方法下。

而調用該方法之后,當前線程會釋放鎖。(提示:這里很重要,也是下文代碼中用 while 而非 if 的原因。)

notify() 方法只能通知一個線程,如果多個線程在等待,那就喚醒任意一個。

notifyAll() 方法是可以喚醒所有等待線程,然后加入同步隊列。

這里我們用到了 2 個隊列:

  • 同步隊列:對應于我們上一節講的線程狀態中的 Runnable,也就是線程準備就緒,就等著搶資源了。
  • 等待隊列:對應于我們上一節講的線程狀態中的 Waiting,也就是等待狀態。

這里需要注意,從等待狀態線程無法直接進入 Q2,而是要先重新加入同步隊列,再次等待拿鎖,拿到了鎖才能進去 Q2;一旦出了 Q2,鎖就丟了。

在 Q2 里,其實只有一個線程,因為這里我們必須要加鎖才能進行操作。

實現

這里我首先建了一個簡單的 Product 類,用來表示生產和消費的產品,大家可以自行添加更多的 fields。

public class Product {private String name;public Product(String name) {this.name = name;}public String getName() {return name;}public void setName(String name) {this.name = name;} }

主函數里我設定了兩類線程,并且這里選擇用普通的 ArrayDeque 來實現 Queue,更簡單的方式是直接用 Java 中的 BlockingQueue 來實現。

BlockingQueue 是阻塞隊列,它有一系列的方法可以讓線程實現自動阻塞,常用的 BlockingQueue 有很多,后面會單獨出一篇文章來講。

這里為了更好的理解并發協同的這個過程,我們先自己處理。

public class Test {public static void main(String[] args) {Queue<Product> queue = new ArrayDeque<>();for (int i = 0; i < 100; i++) {new Thread(new Producer(queue, 100)).start();new Thread(new Consumer(queue, 100)).start();}} }

然后就是 Producer 和 Consumer 了。

public class Producer implements Runnable{private Queue<Product> queue;private int maxCapacity;public Producer(Queue queue, int maxCapacity) {this.queue = queue;this.maxCapacity = maxCapacity;}@Overridepublic void run() {synchronized (queue) {while (queue.size() == maxCapacity) { //一定要用 while,而不是 if,下文解釋try {System.out.println("生產者" + Thread.currentThread().getName() + "等待中... Queue 已達到最大容量,無法生產");wait();System.out.println("生產者" + Thread.currentThread().getName() + "退出等待");} catch (InterruptedException e) {e.printStackTrace();}}if (queue.size() == 0) { //隊列里的產品從無到有,需要通知在等待的消費者queue.notifyAll();}Random random = new Random();Integer i = random.nextInt();queue.offer(new Product("產品" + i.toString()));System.out.println("生產者" + Thread.currentThread().getName() + "生產了產品:" + i.toString());}} }

其實它的主邏輯很簡單,我這里為了方便演示加了很多打印語句才顯得有點復雜。

我們把主要邏輯拎出來看:

public void run() {synchronized (queue) {while (queue.size() == maxCapacity) { //一定要用 while,而不是 if,下文解釋try {wait();} catch (InterruptedException e) {e.printStackTrace();}}if (queue.size() == 0) {queue.notifyAll();}queue.offer(new Product("產品" + i.toString()));}} }

這里有 3 塊內容,再對照這個過程來看:

  • 生產者線程拿到鎖后,其實就是進入了 Q2 階段。首先檢查隊列是否容量已滿,如果滿了,那就要去 Q3 等待;
  • 如果不滿,先檢查一下隊列原本是否為空,如果原來是空的,那就需要通知消費者;
  • 最后生產產品。
  • 這里有個問題,為什么只能用 while 而不是 if?

    其實在這一小段,生產者線程經歷了幾個過程:

  • 如果隊列已滿,它就沒法生產,那也不能占著位置不做事,所以要把鎖讓出來,去 Q3 - 等待隊列 等著;
  • 在等待隊列里被喚醒之后,不能直接奪過鎖來,而是要先加入 Q1 - 同步隊列 等待資源;
  • 一旦搶到資源,關門上鎖,才能來到 Q2 繼續執行 wait() 之后的活,但是,此時這個隊列有可能又滿了,所以退出 wait() 之后,還需要再次檢查 queue.size() == maxCapacity 這個條件,所以要用 while。
  • 那么為什么可能又滿了呢?

    因為線程沒有一直拿著鎖,在被喚醒之后,到拿到鎖之間的這段時間里,有可能其他的生產者線程先拿到了鎖進行了生產,所以隊列又經歷了一個從不滿到滿的過程。

    總結:在使用線程的等待通知機制時,一般都要在 while 循環中調用 wait() 方法。

    消費者線程是完全對稱的,我們來看代碼。

    public class Consumer implements Runnable{private Queue<Product> queue;private int maxCapacity;public Consumer(Queue queue, int maxCapacity) {this.queue = queue;this.maxCapacity = maxCapacity;}@Overridepublic void run() {synchronized (queue) {while (queue.isEmpty()) {try {System.out.println("消費者" + Thread.currentThread().getName() + "等待中... Queue 已缺貨,無法消費");wait();System.out.println("消費者" + Thread.currentThread().getName() + "退出等待");} catch (InterruptedException e) {e.printStackTrace();}}if (queue.size() == maxCapacity) {queue.notifyAll();}Product product = queue.poll();System.out.println("消費者" + Thread.currentThread().getName() + "消費了:" + product.getName());}} }

    結果如下:

    小結

    生產者 - 消費者問題是面試中經常會遇到的題目,本文首先講了該模型的三大優點:解藕,異步,平衡速度差異,然后講解了等待/通知的消息機制以及在該模型中的應用,最后進行了代碼實現。

    文中所有代碼已經放到了我的 Github 上:https://github.com/xiaoqi6666/NYCSDE。

    這個 Github 匯總了我所有的文章和資料,之后也會一直更新和維護,還希望大家幫忙點個 Star,你們的支持和認可,就是我創作的最大動力,我們下篇文章見!

    我是小齊,紐約程序媛,終生學習者,每天晚上 9 點,云自習室里不見不散!

    總結

    以上是生活随笔為你收集整理的多生产者_多线程必考的「生产者 - 消费者」模型,看齐姐这篇文章就够了的全部內容,希望文章能夠幫你解決所遇到的問題。

    如果覺得生活随笔網站內容還不錯,歡迎將生活随笔推薦給好友。