生產消費者模式實現方式和線程安全問題代碼示例

2024-07-13 10:15:40

字體：大中小

來源：轉載

供稿：網友

生產者消費者模式的幾種實現方式

拿我們生活中的例子來說，工廠生產出來的產品總是要輸出到外面使用的，這就是生產與消費的概念。

在我們實際的軟件開發過程中，經常會碰到如下場景：某個模塊負責產生數據，這些數據由另一個模塊來負責處理(此處的模塊是廣義的，可以是類、函數、線程、進程等)。

產生數據的模塊，就形象地稱為生產者;而處理數據的模塊，就稱為消費者。

第一種:采用wait—notify實現生產者消費者模式

1. 一生產者與一消費者：

2. 一生產者與多消費者：

第二種: 采用阻塞隊列實現生產者消費者模式

3. 使用阻塞隊列實現生產者消費者模式

相信大家都有去吃過日本料理。有個很誘人的餐食就是烤肉，烤肉師父會站在一邊一直烤肉，再將烤好的肉放在一個盤子中;而流著口水的我們這些食客會坐在一邊，只要盤子里有肉我們就會一直去吃。

在這個生活案例中，烤肉師父就是生產者，他就負責烤肉，烤完了就把肉放在盤子里，而不是直接遞給食客(即不用通知食客去吃肉)，如果盤子肉滿，師父就會停一會，直到有人去食用烤肉后再去進行生產肉;而食客的我們就只是盯著盤子，一旦盤子有肉我們就負責去吃就行;

整個過程中食客與烤肉師父都不是直接打交道的，而是都與盤子進行交互。

盤子充當了一個緩沖區的概念，有東西生產出來就把東西放進去，盤子也是有大小限制，超過盤子大小就會阻塞生產者生產，等待消費者去消費;當盤子為空的時候，即阻塞消費者消費，等待生產者去生產。

編程中阻塞隊列即可以實現盤子這個功能。

阻塞隊列的特點：

當隊列元素已滿的時候，阻塞插入操作;

當隊列元素為空的時候，阻塞獲取操作。

ArrayBlockingQueue 與 LinkedBlockingQueue都是支持FIFO(先進先出)，但是LinkedBlockingQueue是無界的，而ArrayBlockingQueue 是有界的。

下面使用阻塞隊列實現生產者消費者：

生產者：

import java.util.concurrent.BlockingQueue;public class Producer implements Runnable{	private final BlockingQueue blockingQueue;	//設置隊列緩存的大小。生產過程中超過這個大小就暫時停止生產	private final int QUEUE_SIZE = 10;	public Producer(BlockingQueue blockingQueue){		this.blockingQueue = blockingQueue;	}	int task = 1;	@Override	  public void run() {		while(true){			try {				System.out.println("正在生產：" + task);				//將生產出來的產品放在隊列緩存中				blockingQueue.put(task);				++task;				//讓其停止一會，便于查看效果				Thread.sleep(1000);			}			catch (InterruptedException e) {				e.printStackTrace();			}		}	}}

消費者：

import java.util.concurrent.BlockingQueue;//消費者public class Consumer implements Runnable{	private final BlockingQueue blockingQueue;	public Consumer(BlockingQueue blockingQueue){		this.blockingQueue = blockingQueue;	}	@Override	  public void run() {		//只要阻塞隊列中有任務，就一直去消費		while(true){			try {				System.out.println("正在消費： " + blockingQueue.take());				//讓其停止一會，便于查看效果				Thread.sleep(2000);			}			catch (InterruptedException e) {				e.printStackTrace();			}		}	}}

測試：

import java.util.concurrent.BlockingQueue;import java.util.concurrent.LinkedBlockingQueue;/** * 生產者消費者模式 * 使用阻塞隊列BlockingQueue * @author wanggenshen * */public class TestConPro {	public static void main(String[] args){		BlockingQueue blockingQueue = new LinkedBlockingQueue(5);		Producer p = new Producer(blockingQueue);		Consumer c = new Consumer(blockingQueue);		Thread tp = new Thread(p);		Thread tc= new Thread(c);		tp.start();		tc.start();	}}

因為LinkedBlockingQueue是無界隊列，所以生產者會不斷去生產，將生產出的任務放在隊列中，消費者去隊列中去消費：

生產者消費者模式代碼,線程安全代碼,線程安全的代碼

如果改用有界阻塞隊列ArrayBlockingQueue，就可以初始化隊列的大小。則隊列中元素超過隊列大小的時候，生產者就會等待消費者消費一個再去生產一個：

測試代碼：

初始化一個大小為10的ArrayBlockingQueue：

public static void main(String[] args){	BlockingQueue blockingQueue = new ArrayBlockingQueue(10);	Producer p = new Producer(blockingQueue);	Consumer c = new Consumer(blockingQueue);	Thread tp = new Thread(p);	Thread tc= new Thread(c);	tp.start();	tc.start();}

測試中，讓生產者生產速度略快，而消費者速度略慢一點。可以看到生產出來的產品序號與消費的產品序號差始終為10(隊列的大小)：

生產者消費者模式代碼,線程安全代碼,線程安全的代碼