淺談生產者消費者模型(Linux系統下的兩種實現方法)

2024-09-05 23:03:28

字體：大中小

來源：轉載

供稿：網友

生產者消費者問題是同步問題中的一種常見情況，借用一下維基百科的話

生產者消費者問題（英語：Producer-consumer problem），也稱有限緩沖問題（英語：Bounded-buffer problem），是一個多線程同步問題的經典案例。該問題描述了兩個共享固定大小緩沖區的線程——即所謂的“生產者”和“消費者”——在實際運行時會發生的問題。生產者的主要作用是生成一定量的數據放到緩沖區中，然后重復此過程。與此同時，消費者也在緩沖區消耗這些數據。該問題的關鍵就是要保證生產者不會在緩沖區滿時加入數據，消費者也不會在緩沖區中空時消耗數據。

第一種實現信號量配合互斥鎖實現，這種方法很清晰簡單

信號量：

信號量的特性如下：信號量是一個非負整數（車位數），所有通過它的線程/進程（車輛）都會將該整數減一（通過它當然是為了使用資源），當該整數值為零時，所有試圖通過它的線程都將處于等待狀態。在信號量上我們定義兩種操作： Wait（等待）和 Release（釋放）。當一個線程調用Wait操作時，它要么得到資源然后將信號量減一，要么一直等下去（指放入阻塞隊列），直到信號量大于等于一時。Release（釋放）實際上是在信號量上執行加操作，對應于車輛離開停車場，該操作之所以叫做“釋放”是因為釋放了由信號量守護的資源。

wait, release在Linux下

int sem_wait(sem_t * sem);
int sem_post(sem_t * sem);

設定兩個信號量，empty用來表示空槽的個數，full用來表示占有的個數

生產者在向任務隊列里放資源時，調用sem_wait(&full)來檢查隊列是否已滿，如果滿的話，就阻塞，直到有消費者從里面取資源再蘇醒，如果不滿，就放資源，并通知消費者來取。

消費者在從任務隊列里取資源時，調用sem_wait(&empty)來檢查隊列是否為空，如果空的話，就阻塞，直到有生產者向里面放資源再蘇醒，如果不空，就取資源，并通知生產者來放。

而互斥鎖僅僅是為了防止多個線程同時對隊列進行操作，造成未知的結果。

#include <stdio.h>#include <pthread.h>#include <semaphore.h>#define MAX 5 //隊列長度pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;sem_t full; 	//填充的個數sem_t empty; 	//空槽的個數int top = 0;   //隊尾int bottom = 0; //隊頭void* produce(void* arg){	int i;	for ( i = 0; i < MAX*2; i++)	{		printf("producer is preparing data/n");		sem_wait(&empty);//若空槽個數低于0阻塞				pthread_mutex_lock(&mutex);				top = (top+1) % MAX;		printf("now top is %d/n", top);		pthread_mutex_unlock(&mutex);				sem_post(&full);	}	return (void*)1;}void* consume(void* arg){	int i;	for ( i = 0; i < MAX*2; i++)	{		printf("consumer is preparing data/n");		sem_wait(&full);//若填充個數低于0阻塞			pthread_mutex_lock(&mutex);				bottom = (bottom+1) % MAX;		printf("now bottom is %d/n", bottom);		pthread_mutex_unlock(&mutex);				sem_post(&empty);	}	return (void*)2;}int main(int argc, char *argv[]){	pthread_t thid1;	pthread_t thid2;	pthread_t thid3;	pthread_t thid4;	int ret1;	int ret2;	int ret3;	int ret4;	sem_init(&full, 0, 0);	sem_init(&empty, 0, MAX);	pthread_create(&thid1, NULL, produce, NULL);	pthread_create(&thid2, NULL, consume, NULL);	pthread_create(&thid3, NULL, produce, NULL);	pthread_create(&thid4, NULL, consume, NULL);	pthread_join(thid1, (void**)&ret1);	pthread_join(thid2, (void**)&ret2);	pthread_join(thid3, (void**)&ret3);	pthread_join(thid4, (void**)&ret4);	return 0;}

注：如果把sem_wait()和sem_post()放到pthread_mutex_lock()與pthread_mutex_unlock()之間會如何呢？

答案是：死鎖，因為我們不能預知線程進入共享區順序，如果消費者線程先對mutex加鎖，并進入，sem_wait()發現隊列為空，阻塞，而生產者在對mutex加鎖時，發現已上鎖也阻塞，雙方永遠無法喚醒對方。

第二種是條件變量配合互斥鎖實現

條件變量的常見用法是在不滿足某些條件時，阻塞自己，直到有線程通知自己醒來。

而互斥量在這里的作用依然還是防止多線程對共享資源同時操作，造成未知結果。

生產者消費者的行為與之前相同，只不過原來只調用sem_wait()可以完成兩步，1是檢查條件，2是阻塞，現在條件變量需要我們自己來設定條件(所以說條件變量配合互斥鎖比信號量的功能更強大，因為它可以自定義休眠條件，但是這對使用者的要求也提高了，必須理清邏輯關系避免死鎖)

#include <stdio.h>#include <pthread.h>#define MAX 5pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;pthread_cond_t notfull = PTHREAD_COND_INITIALIZER; 	//是否隊滿pthread_cond_t notempty = PTHREAD_COND_INITIALIZER; 	//是否隊空int top = 0;int bottom = 0;void* produce(void* arg){	int i;	for ( i = 0; i < MAX*2; i++)	{		pthread_mutex_lock(&mutex);		while ((top+1)%MAX == bottom)		{			printf("full! producer is waiting/n");			pthread_cond_wait(¬full, &mutex);//等待隊不滿		}		top = (top+1) % MAX;		printf("now top is %d/n", top);		pthread_cond_signal(¬empty);//發出隊非空的消息		pthread_mutex_unlock(&mutex);	}	return (void*)1;}void* consume(void* arg){	int i;	for ( i = 0; i < MAX*2; i++)	{		pthread_mutex_lock(&mutex);		while ( top%MAX == bottom)		{			printf("empty! consumer is waiting/n");			pthread_cond_wait(¬empty, &mutex);//等待隊不空		}		bottom = (bottom+1) % MAX;		printf("now bottom is %d/n", bottom);		pthread_cond_signal(¬full);//發出隊不滿的消息		pthread_mutex_unlock(&mutex);	}	return (void*)2;}int main(int argc, char *argv[]){	pthread_t thid1;	pthread_t thid2;	pthread_t thid3;	pthread_t thid4;	int ret1;	int ret2;	int ret3;	int ret4;	pthread_create(&thid1, NULL, produce, NULL);	pthread_create(&thid2, NULL, consume, NULL);	pthread_create(&thid3, NULL, produce, NULL);	pthread_create(&thid4, NULL, consume, NULL);	pthread_join(thid1, (void**)&ret1);	pthread_join(thid2, (void**)&ret2);	pthread_join(thid3, (void**)&ret3);	pthread_join(thid4, (void**)&ret4);	return 0;}

注：

為什么信號量在互斥區外，而條件變量在互斥區內呢？

因為互斥鎖本質上是二元信號量，和信號量互斥的原理相同，而且放在互斥區會死鎖，而條件變量是和互斥鎖協同配合的，

我們從pthread_cond_wait()和pthread_cond_signal()的內部實現就可以看出

pthread_cond_wait()是先將互斥鎖解開，并陷入阻塞，直到pthread_signal()發出信號后pthread_cond_wait()再加上鎖，然后退出，可以看到它們在設計時就是為了協同配合，而互斥鎖和信號量都是由Linux下的futex機制實現的，這里就不展開說了

這里貼出了pthread_wait()源碼圖

生產者消費者模型