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

更新時間：2017年01月06日 10:11:16 投稿：jingxian

下面小編就為大家?guī)硪黄獪\談生產者消費者模型(Linux系統(tǒng)下的兩種實現方法)。小編覺得挺不錯的，現在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

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

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

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

信號量：

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

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()之間會如何呢？

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

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

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

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

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

#include <stdio.h>
#include <pthread.h>

#define MAX 5

pthread_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);//發(fā)出隊非空的消息

		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);//發(fā)出隊不滿的消息

		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;
}

注：

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

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

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

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

這里貼出了pthread_wait()源碼圖