快捷導(dǎo)航

淺談生產(chǎn)者消費(fèi)者模型(Linux系統(tǒng)下的兩種實(shí)現(xiàn)方法)

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

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

生產(chǎn)者消費(fèi)者問題是同步問題中的一種常見情況，借用一下維基百科的話

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

第一種實(shí)現(xiàn)信號(hào)量配合互斥鎖實(shí)現(xiàn)，這種方法很清晰簡(jiǎn)單

信號(hào)量：

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

wait, release在Linux下

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

設(shè)定兩個(gè)信號(hào)量，empty用來表示空槽的個(gè)數(shù)，full用來表示占有的個(gè)數(shù)

生產(chǎn)者在向任務(wù)隊(duì)列里放資源時(shí)，調(diào)用sem_wait(&full)來檢查隊(duì)列是否已滿，如果滿的話，就阻塞，直到有消費(fèi)者從里面取資源再蘇醒，如果不滿，就放資源，并通知消費(fèi)者來取。

消費(fèi)者在從任務(wù)隊(duì)列里取資源時(shí)，調(diào)用sem_wait(&empty)來檢查隊(duì)列是否為空，如果空的話，就阻塞，直到有生產(chǎn)者向里面放資源再蘇醒，如果不空，就取資源，并通知生產(chǎn)者來放。

而互斥鎖僅僅是為了防止多個(gè)線程同時(shí)對(duì)隊(duì)列進(jìn)行操作，造成未知的結(jié)果。

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

#define MAX 5 //隊(duì)列長(zhǎng)度

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
sem_t full; 	//填充的個(gè)數(shù)
sem_t empty; 	//空槽的個(gè)數(shù)

int top = 0;   //隊(duì)尾
int bottom = 0; //隊(duì)頭

void* produce(void* arg)
{
	int i;
	for ( i = 0; i < MAX*2; i++)
	{
		printf("producer is preparing data\n");
		sem_wait(&empty);//若空槽個(gè)數(shù)低于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);//若填充個(gè)數(shù)低于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()之間會(huì)如何呢？

答案是：死鎖，因?yàn)槲覀儾荒茴A(yù)知線程進(jìn)入共享區(qū)順序，如果消費(fèi)者線程先對(duì)mutex加鎖，并進(jìn)入，sem_wait()發(fā)現(xiàn)隊(duì)列為空，阻塞，而生產(chǎn)者在對(duì)mutex加鎖時(shí)，發(fā)現(xiàn)已上鎖也阻塞，雙方永遠(yuǎn)無(wú)法喚醒對(duì)方。

第二種是條件變量配合互斥鎖實(shí)現(xiàn)

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

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

生產(chǎn)者消費(fèi)者的行為與之前相同，只不過原來只調(diào)用sem_wait()可以完成兩步，1是檢查條件，2是阻塞，現(xiàn)在條件變量需要我們自己來設(shè)定條件(所以說條件變量配合互斥鎖比信號(hào)量的功能更強(qiáng)大，因?yàn)樗梢宰远x休眠條件，但是這對(duì)使用者的要求也提高了，必須理清邏輯關(guān)系避免死鎖)

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

#define MAX 5

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t notfull = PTHREAD_COND_INITIALIZER; 	//是否隊(duì)滿
pthread_cond_t notempty = PTHREAD_COND_INITIALIZER; 	//是否隊(duì)空

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);//等待隊(duì)不滿
		}

		top = (top+1) % MAX;
		printf("now top is %d\n", top);
		pthread_cond_signal(¬empty);//發(fā)出隊(duì)非空的消息

		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);//等待隊(duì)不空
		}
		bottom = (bottom+1) % MAX;
		printf("now bottom is %d\n", bottom);
		pthread_cond_signal(¬full);//發(fā)出隊(duì)不滿的消息

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

注：

為什么信號(hào)量在互斥區(qū)外，而條件變量在互斥區(qū)內(nèi)呢？

因?yàn)榛コ怄i本質(zhì)上是二元信號(hào)量，和信號(hào)量互斥的原理相同，而且放在互斥區(qū)會(huì)死鎖，而條件變量是和互斥鎖協(xié)同配合的，

我們從pthread_cond_wait()和pthread_cond_signal()的內(nèi)部實(shí)現(xiàn)就可以看出

pthread_cond_wait()是先將互斥鎖解開，并陷入阻塞，直到pthread_signal()發(fā)出信號(hào)后pthread_cond_wait()再加上鎖，然后退出，可以看到它們?cè)谠O(shè)計(jì)時(shí)就是為了協(xié)同配合，而互斥鎖和信號(hào)量都是由Linux下的futex機(jī)制實(shí)現(xiàn)的，這里就不展開說了

這里貼出了pthread_wait()源碼圖