Linux之生產(chǎn)者消費者模型用法解讀
一、生產(chǎn)者消費者模型
舉個例子：
我們現(xiàn)實生活中：工廠，超市，人[消費者]之間的關系就是一個典型的生產(chǎn)者消費者模型
這個模型的基本工作流程是：
- 工廠制造商品供貨給超市，消費者到超市購買商品
現(xiàn)實生活中，為什么會優(yōu)化出這樣的一種模型？
為什么消費者不直接去工廠買東西？
主要有以下3個原因：
為了效率
1.對于工廠來說，消費者如果直接來買商品，一個消費者一次購買的商品數(shù)量非常有限，而且一個工廠的產(chǎn)品種類一般并不多，所以消費者還得去不同工廠買東西，工廠為了防止賣不完，只能減少商品的生產(chǎn)，即降低了工廠的生產(chǎn)效率，但是如果賣給超市，就可以一次性賣一大車貨物，多找?guī)讉€超市，就可以做到：生產(chǎn)多少賣多少了
2.對于消費者，工廠的占地面積比較大，所以一般都距離居民地比較遠，而超市一般都比工廠占地面積小的多，離消費者很近，所以消費者去超市買東西比去工廠更快，而且一個工廠的產(chǎn)品種類一般并不多，但是超市售賣的產(chǎn)品種類繁多
有了超市就可以做到生產(chǎn)者和消費者之間的解藕
1.一個工廠關門了，超市換一個工廠進貨就行了，只要進的貨還是一樣的，就絲毫不影響消費者消費。事實就是如此，我們消費者從超市買東西時，根本就不知道這個商品是哪個工廠生產(chǎn)的，我們也不關心消費者只管購買商品就行了
2.到超市消費者是誰，是什么群體，工廠也不知道，也不關心，這個是超市該關心的，因為工廠只把商品賣給超市，超市賣給誰和工廠無關，工廠只管生產(chǎn)就行了
3.有了超市的存在
- 左側(cè)的工廠發(fā)生變化不影響右側(cè)的消費者
- 右側(cè)的消費者發(fā)生變化不影響左側(cè)的工廠
- 這不就
解藕了嗎？
支持忙閑不均
有了超市這個緩沖區(qū)的存在
1.在購物潮之前，超市就可以通知工廠多生產(chǎn)一些商品，超市多進貨，方便應對更多消費者
2.在囤積的商品多的時候，超市也可以搞活動吸引消費者，并且通知工廠生產(chǎn)地慢一點
超市本質(zhì)上就是工廠和消費者之間的緩存
即：超市支持工廠“預加載”產(chǎn)品，消費者可以一定程度“預定”產(chǎn)品
上述例子對應到計算機中的生產(chǎn)者消費者模型就是：
- ①工廠：生產(chǎn)者線程
- ②消費者：消費者線程
- ③超市：以某種數(shù)據(jù)結(jié)構(gòu)組織的內(nèi)存區(qū)域
- ④商品：數(shù)據(jù)
和線程安全再對應一下：
- ①超市：
共享/臨界資源 - ②我們要研究生產(chǎn)者消費者模型，就需要研究清楚多個生產(chǎn)者和多個消費者之間的同步互斥關系！
一共有3種關系：
1.生產(chǎn)者和生產(chǎn)者之間：互斥
- 因為如果多個生產(chǎn)者同時向超市（共享資源）寫數(shù)據(jù)，可能會產(chǎn)生線程安全問題
- 因為同時有多個線程對同一份共享資源進行修改，很容易互相覆蓋，出現(xiàn)并發(fā)線程安全問題 即：線程之間切換時可能導致數(shù)據(jù)不一致問題
- 其次共享資源的空間就那么大，多個生產(chǎn)者線程同時去寫的話，一個線程寫的多了，另一個線程就只能少寫一點了
- 或者線程a剛向共享資源中的一個位置寫了一個1，線程b就跑過來把那個位置的1改成10了
- 所以生產(chǎn)者線程之間是競爭關系
2.消費者和消費者之間：互斥
- 消費者線程，雖然是進入臨界資源讀取數(shù)據(jù)，但是其實也會對共享資源進行修改
- 即：線程拿走了一個數(shù)據(jù)a，其他線程看待這個這個數(shù)據(jù)a就是過期數(shù)據(jù)了（和到超市買東西一樣，買了一包方便面，超市就少了一包方便面）
- 既然會修改，那么多個消費者線程同時進入共享資源進行修改的話，也可能會出現(xiàn)并發(fā)切換導致的線程安全問題
- 比如：線程a和線程b同時訪問共享資源，線程a先把數(shù)據(jù)X消費走了，但是因為是同時進入線程b任認為數(shù)據(jù)X還是有效的，也拿了數(shù)據(jù)X（如果共享資源是數(shù)組，那可以把數(shù)據(jù)X理解為數(shù)組中的一個元素）
- 就會導致一份數(shù)據(jù)，被使用了兩次
3.生產(chǎn)者和消費者之間：互斥+同步
- 因為不管是生產(chǎn)者線程還是消費者線程，訪問共享資源時，都會進行修改，多線程并發(fā)修改共享資源，就可能會出現(xiàn)并發(fā)線程安全問題
- 所以它們首先得是
互斥的 - 但是如果只有互斥，那么消費者如果不知道有沒有數(shù)據(jù)，就只能不斷地去輪詢檢測，每次輪詢都要申請鎖，那生產(chǎn)者線程就可能很難搶到鎖，就一直生產(chǎn)不了數(shù)據(jù)，消費者線程就一直拿不到數(shù)據(jù)，造成惡性循環(huán)
- 就可能會導致鎖的饑餓問題
- 所以需要
同步關系來提高生產(chǎn)者和消費者模型的效率 即： - 設置條件變量，讓消費者線程在條件變量的等待隊列中等，當生產(chǎn)者線程生產(chǎn)數(shù)據(jù)之后，才喚醒消費者線程，去讀取數(shù)據(jù)
二、生產(chǎn)者消費者模型的阻塞隊列版本
生產(chǎn)者消費者模型一般會使用一個阻塞隊列來作為共享資源，進而實現(xiàn)多線程協(xié)作
生產(chǎn)者消費者模型的阻塞隊列的特點：
- 如果隊列為空，那么一個消費者線程如果來拿數(shù)據(jù)，它就會被阻塞
- 如果隊列為滿，那么一個生產(chǎn)者線程如果還要向隊列里寫數(shù)據(jù)，它就會被阻塞
- 如果隊列不空也不滿，那么生產(chǎn)者就可以向隊列尾部寫數(shù)據(jù)，消費者就可以向隊列頭部拿數(shù)據(jù)
阻塞隊列類的簡單實現(xiàn)
成員變量
1.存儲數(shù)據(jù)的容器
直接使用STL的queue，因為數(shù)據(jù)的類型不確定，所以阻塞隊列類是模板類
2.一把鎖
阻塞隊列自己會被所有線程看見，所以它是共享資源，所以需要鎖來保護自己
要幾把鎖呢？
因為所有生產(chǎn)者線程之間，所有消費者線程之間，以及生產(chǎn)者和消費者之間
都是互斥的，所以它們得用同一把鎖來實現(xiàn)互斥
3.生產(chǎn)者線程的條件變量
因為在滿足一定條件（比如：阻塞隊列滿了，或者阻塞隊列滿了4/5了等）時，可以讓所有生產(chǎn)者線程暫時暫停生產(chǎn)（即去條件變量的等待隊列中阻塞）
4.消費者線程的條件變量
因為在滿足一定條件（比如：阻塞隊列為空，或者阻塞隊列空了4/5了等）時，可以讓所有消費者線程暫時暫停消費（即去條件變量的等待隊列中阻塞）
為什么要搞兩個條件變量？
一個條件變量雖然也可以實現(xiàn)生產(chǎn)者線程和消費者線程之間的同步
但是實現(xiàn)起來非常麻煩，而且不能區(qū)分條件變量的等待隊列下的是生產(chǎn)者線程還是消費者線程
而且
兩個條件變量可以很好地支持：
生產(chǎn)者消費者模型的第3個優(yōu)點：忙閑不均
5.int _cap：阻塞隊列的最大容量
6.int _csleep_num：在消費者條件變量的等待隊列中等待的線程個數(shù)
7.int _psleep_num：在生產(chǎn)者條件變量的等待隊列中等待的線程個數(shù)
6和7成員變量的存在主要是為了方便實現(xiàn)線程之間的互相喚醒機制
（即生產(chǎn)者線程生產(chǎn)了之后，可以喚醒消費者線程來消費，反之同理）
成員函數(shù)
- Equeue：生產(chǎn)數(shù)據(jù)
void Equeue(const T& in)
{
pthread_mutex_lock(&_mutex);
//生產(chǎn)者調(diào)用
while(IsFull())
{
_psleep_num++;
cout << "生產(chǎn)者， 進入休眠了：" << _psleep_num << endl;
pthread_cond_wait(&_full_cond, &_mutex);
_psleep_num--;
}
//100% 隊列有空間
_q.push(in);
if(_csleep_num > 0)
{
pthread_cond_signal(&_empty_cond);
cout << "喚醒消費者..." << endl;
}
pthread_mutex_unlock(&_mutex);
}
代碼細節(jié)：
偽喚醒問題的解決
- 即：判斷線程是否要進入條件變量的等待隊列時，判斷不能用if而要用while
- 不然就有可能出現(xiàn)偽喚醒問題：即在條件變量下等待的線程，喚醒條件其實并不滿足
- 但是因為程序員編碼的問題，可能意外被喚醒了
例如：
- 生產(chǎn)者消費者模型中，因為阻塞隊列中沒有數(shù)據(jù)，所以全部都5個消費者線程在條件變量的等待隊列中等待
- 生產(chǎn)者線程生產(chǎn)了一個數(shù)據(jù)，意外地把喚醒了多個消費者線程
- 然后一個消費者線程搶到鎖之后，把阻塞隊列中那唯一的一個數(shù)據(jù)搶走了，它解鎖之后
- 因為喚醒了多個消費者線程
- 所以鎖可能又被一個消費者線程搶到了，但是此時阻塞隊列中根本沒有數(shù)據(jù)！
此時：
1.如果此時是使用if進行“線程是否需要進入條件變量的等待隊列"的判斷的這個被偽喚醒的線程，重新申請并拿到鎖之后，就直接"餓虎出籠"去肆意妄為了
2.如果是使用while進行“線程是否需要進入條件變量的等待隊列”的判斷的，這個被喚醒的線程，重新申請并拿到鎖之后，也還是不能直接出循環(huán)，因為要再判斷一下循環(huán)條件是否不滿足了
雖然循環(huán)條件是"線程需要進入等待隊列"的條件，但是如果這個條件滿足，不就意味著線程不應該被喚醒嗎？
- Pop：獲取并刪除數(shù)據(jù)
T Pop()
{
//消費者調(diào)用
pthread_mutex_lock(&_mutex);
while(IsEmpty())
{
_csleep_num++;
pthread_cond_wait(&_empty_cond, &_mutex);
_csleep_num--;
}
T data = _q.front();
_q.pop();
if(_psleep_num > 0)
{
pthread_cond_signal(&_full_cond);
cout << "喚醒生產(chǎn)者.." << endl;
}
pthread_mutex_unlock(&_mutex);
return data;
}
- IsEmpty：阻塞隊列是否為空
- IsFull：阻塞隊列是否為滿
bool IsFull()
{
return _q.size() >= _cap;
}
bool IsEmpty()
{
return _q.empty();
}
源碼
#pragma once
#include <iostream>
#include <pthread.h>
#include <queue>
#include <string>
using namespace std;
int defalutcap = 5;
template<typename T>
class BlockQueue
{
private:
bool IsFull()
{
return _q.size() >= _cap;
}
bool IsEmpty()
{
return _q.empty();
}
public:
BlockQueue(int cap = defalutcap)
: _cap(cap),
_csleep_num(0),
_psleep_num(0)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_full_cond, nullptr);
pthread_cond_init(&_empty_cond, nullptr);
}
void Equeue(const T& in)
{
pthread_mutex_lock(&_mutex);
//生產(chǎn)者調(diào)用
while(IsFull())
{
_psleep_num++;
cout << "生產(chǎn)者， 進入休眠了：" << _psleep_num << endl;
pthread_cond_wait(&_full_cond, &_mutex);
_psleep_num--;
}
//100% 隊列有空間
_q.push(in);
if(_csleep_num > 0)
{
pthread_cond_signal(&_empty_cond);
cout << "喚醒消費者..." << endl;
}
pthread_mutex_unlock(&_mutex);
}
T Pop()
{
//消費者調(diào)用
pthread_mutex_lock(&_mutex);
while(IsEmpty())
{
_csleep_num++;
pthread_cond_wait(&_empty_cond, &_mutex);
_csleep_num--;
}
T data = _q.front();
_q.pop();
if(_psleep_num > 0)
{
pthread_cond_signal(&_full_cond);
cout << "喚醒生產(chǎn)者.." << endl;
}
pthread_mutex_unlock(&_mutex);
return data;
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_full_cond);
pthread_cond_destroy(&_empty_cond);
}
private:
//臨界資源
queue<T> _q;
//大小
int _cap;
pthread_mutex_t _mutex;
pthread_cond_t _full_cond;
pthread_cond_t _empty_cond;
int _csleep_num;//消費者休眠的個數(shù)
int _psleep_num;//生產(chǎn)者休眠的個數(shù)
};
三、POSIX信號量
POSIX信號量和SystemV信號量作?相同，都是?于同步操作，達到?沖突的訪問共享資源?的。但POSIX可以?于線程間同步。
初始化信號量
sem_init
作用: 用于初始化一個未命名的POSIX信號量（也稱為匿名信號量），通常用于線程間同步或共享內(nèi)存的進程間同步。
#include <semaphore.h> int sem_init(sem_t *sem, int pshared, unsigned int value);
- sem_t *sem: 指向要初始化的信號量對象的指針。
- pshared: 0表?線程間共享，?零表?進程間共享
- value：信號量初始值
返回值
- 成功：返回 0
- 失?。悍橇?/li>
銷毀信號量
sem_destroy
int sem_destroy(sem_t *sem);
等待信號量
sem_wait
是 POSIX 信號量的 P操作（等待/獲取信號量），用于對信號量進行原子減1操作。
它的主要作用是：
- 如果信號量值 > 0：立即將其減 1，線程繼續(xù)執(zhí)行。
- 如果信號量值 = 0：線程阻塞，直到信號量值變?yōu)檎龜?shù)（其他線程或進程調(diào)用 sem_post 釋放資源）。
int sem_wait(sem_t *sem); //P()
發(fā)布信號量
sem_post
是 POSIX 信號量的 V操作（釋放/增加信號量），用于對信號量進行原子加1操作。它的主要作用是：
- 將信號量的值 +1，表示釋放一個資源。
- 如果有線程阻塞在 sem_wait，則喚醒其中一個線程（取決于系統(tǒng)調(diào)度策略）。
int sem_post(sem_t *sem);//V()
四、基于環(huán)形隊列的?產(chǎn)消費模型
上?節(jié)?產(chǎn)者-消費者的例?是基于queue的,其空間可以動態(tài)分配,現(xiàn)在基于固定??的環(huán)形隊列重寫這個程序（POSIX信號量）:
Sem的封裝
#include <iostream>
#include <semaphore.h>
#include <pthread.h>
namespace SemMoudle
{
const int defaultvalue = 1;
class Sem
{
public:
Sem(unsigned int sem_vlaue = defaultvalue)
{
sem_init(&_sem, 0, sem_vlaue);
}
void P()
{
//等待信號量，會將信號量的值減1
int n = sem_wait(&_sem);//原子的
(void)n;
}
void V()
{
//發(fā)布信號量，釋放資源，會將信號量的值加1
int n = sem_post(&_sem);//原子的
(void)n;
}
~Sem()
{
sem_destroy(&_sem);
}
private:
sem_t _sem;
};
}
Mutex的封裝
#pragma once
#include <iostream>
#include <pthread.h>
namespace MutexModue
{
class Mutex
{
public:
Mutex()
{
pthread_mutex_init(&_mutex, nullptr);
}
void Lock()
{
int n = pthread_mutex_lock(&_mutex);
(void)n;
}
void Unlock()
{
int n = pthread_mutex_unlock(&_mutex);
(void)n;
}
~Mutex()
{
pthread_mutex_destroy(&_mutex);
}
pthread_mutex_t* Get()
{
return &_mutex;
}
private:
pthread_mutex_t _mutex;
};
class LockGuard
{
public:
LockGuard(Mutex& mutex):_mutex(mutex)
{
_mutex.Lock();
}
~LockGuard()
{
_mutex.Unlock();
}
private:
Mutex& _mutex;
};
}
環(huán)形隊列
#pragma once
#include <iostream>
#include <vector>
#include "Sem.hpp"
#include "Mutex.hpp"
using namespace std;
static const int gcap = 5;
using namespace SemMoudle;
using namespace MutexModue;
//環(huán)形隊列
template<typename T>
class RingQueue
{
public:
RingQueue(int cap = gcap)
:_cap(cap),
_rq(cap),
_blank_sem(cap),
_p_step(0),
_data_sem(0),
_c_step(0)
{}
void Equeue(const T& in)
{
//生產(chǎn)者
//1.申請信號量，空位置信號量
_blank_sem.P();
{
LockGuard lockguard(_pmutex);
//2.生產(chǎn)
_rq[_p_step] = in;
//3.更新下標
++_p_step;
//4.維持環(huán)形特性
_p_step %= _cap;
}
_data_sem.V();
}
void Pop(T* out)
{
//消費者
//1.申請信號量，數(shù)據(jù)信號量
_data_sem.P();
{
LockGuard lockguard(_cmutex);
//2.消費
*out = _rq[_c_step];
//3.更新下標
++_c_step;
//4.維持環(huán)形特性
_c_step %= _cap;
}
_blank_sem.V();
}
private:
vector<T> _rq;
int _cap;
//生產(chǎn)者
Sem _blank_sem;//空位置
int _p_step;
//消費者
Sem _data_sem;//數(shù)據(jù)
int _c_step;
//維護多生產(chǎn)，多消費，2把鎖
Mutex _cmutex;
Mutex _pmutex;
};
總結(jié)
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
Linux中創(chuàng)建,復制和刪除文件及目錄的命令詳解
這篇文章主要為大家詳細介紹了Linux中創(chuàng)建,復制和刪除文件及目錄的相關命令,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下2026-02-02
下面小編就為大家?guī)硪黄猅hrift的安裝方法和簡單實例。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2016-12-12
Linux?Docker安裝Jenkins并實現(xiàn)Maven工程自動化部署過程
這段文章詳細介紹了在Linux系統(tǒng)上安裝和配置Jenkins的過程,包括關閉防火墻、訪問Jenkins界面、解鎖Jenkins容器、安裝Maven插件、配置Jenkins連接碼云倉庫以及設置構(gòu)建后操作等步驟2026-06-06
CentOS 7中 Minimal 安裝JDK 1.8的教程
這篇文章主要介紹了CentOS 7 Minimal 安裝JDK 1.8的教程，非常不錯，具有參考借鑒價值 ,需要的朋友可以參考下2018-05-05
這篇文章主要為大家詳細介紹了Centos6下root密碼忘記的解決辦法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-08-08
這篇文章主要介紹了centos7下如何安裝ftp服務問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
使用Linux的read和write系統(tǒng)函數(shù)操作文件的方法詳解
在Linux系統(tǒng)編程中,文件操作是非常基礎且重要的部分,Linux提供了多個系統(tǒng)調(diào)用來實現(xiàn)文件的讀寫操作,其中read和write是最常用的兩個函數(shù),本文將詳細介紹這兩個系統(tǒng)調(diào)用的功能、使用方法以及實際應用中的注意事項,需要的朋友可以參考下2025-10-10
在Linux中,diff命令是一種強大的工具,用于比較兩個文本文件的差異,除了基本的用法外,還提供了許多選項來滿足不同的需求,如忽略大小寫、空格變化等,此外,還可以使用colordiff和vimdiff等高級工具來增強對比的可讀性,為了忽略行順序的差異,可以先對文件進行排序2025-12-12
httpd.conf下讓apache 支持 php5 的配置2010-03-03
如果你想知道你的服務器正在做干什么，你就需要了解一些基本的命令，一旦你精通了這些命令，那你就是一個 專業(yè)的 Linux 系統(tǒng)管理員2012-03-03