Linux--多執行緒（三）

生產者消費者模型

概念： 生產者消費者模式就是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過一個來進行通訊，所以生產者生產完數據之後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列里取，阻塞隊列就相當於一個緩衝區，平衡了生產者和消費者的處理能力。這個阻塞隊列就是用來給生產者和消費者解耦的。

生產消費者模型優點：

  • 解耦：生產者和消費者是通過一個共享數據區域來進行通訊。而不是直接進行通訊，這樣兩個角色之間的依耐性就降低了（程式碼層面實現解耦），變成了角色與共享數據區域之間的弱耦合，一個邏輯出錯不影響兩一個邏輯，二者變得更獨立。
  • 支援並發：生產者負責生產數據，消費者負責拿數據。生產者生產完數據可以繼續生產，大部分時間內是不需要等待消費者消費數據才繼續生產。也就是說，在任一時刻，二者都是在正常處理任務的，進度都得以推進。
  • 支援忙閑下不均：生產者生產了數據是放進容器中，消費者不必立即消費，可以慢慢地從容器中取數據。容器快要空了，消費者的消費速度就可以降下來，讓生產者繼續生產。

生產消費模型特徵（簡記321）：

  1. 3種關係： 生產者與生產者（互斥）、生產者與消費者（同步（主要）和互斥）和消費者與消費者（互斥）
  2. 兩個角色： 生產者和消費者
  3. 一個交易場所： 容器、共享資源等

基於阻塞隊列的生產者消費者模型

阻塞隊列的特點：

  • 當隊列為空時，從隊列獲取元素的操作將會被阻塞，直到隊列中被放入了元素
  • 當隊列滿時，往隊列里存放元素的操作也會被阻塞，直到有元素被從隊列中取出(以上的操作都是基於不同的執行緒來說的，執行緒在對阻塞隊列進程操作時會被阻塞)
  • 在這個模型中，生產者和消費者的交易場所就是阻塞隊列

實現

概述：

因為交易場所是一個阻塞隊列，所以，我封裝了一個BlcokingQueue 的類，這裡提供了放數據和取數據這樣兩個主要的方法。其中，有五個成員變數：

  • 隊列： 使用STL中的queue來實現
  • 容量： 阻塞隊列的容量，由用戶給定，我們也可以提供一個默認的容量
  • 互斥量： 為了實現生產者和消費者的同步，我們需要使用條件變數和互斥量來實現同步的操作
  • 生產者喚醒和等待的條件變數： 當隊列滿了，生產者等待條件滿足，應該掛起等待，等待消費者喚醒
  • 消費者喚醒和等待的條件變數： 當隊列為空，消費者等待條件滿足，應該掛起等待，等待生產者喚醒

構造函數和析構函數：

  • 構造函數做初始化和資源分配的操作，分配鎖資源和條件變數

  • 析構函數做清理資源的操作，對鎖和條件變數進行銷毀

隊列類的整體架構

#include<iostream>
#include<string.h>
using namespace std;//標準命名空間
//類模板
template<class T>
class BlockQueue
{
public:
    //構造函數，化容量為5
    BlockQueue(int capacity = 5)
    :_capacity(capacity)
    {
      //初始化鎖和互斥量
      pthread_mutex_init(&_lock,nullptr);
      pthread_cond_init(&_c_cond,nullptr);
      pthread_cond_init(&_p_cond,nullptr);
    }
    //析構函數
    ~BlockQueue()
    {
      pthread_mutex_destroy(&_lock);
      pthread_cond_destroy(&_c_cond);
      pthread_cond_destroy(&_p_cond);
    }
private:
    //隊列容器
    queue<T> _q;
    size_t   _capacity;//隊列最大容器
    pthread_mutex_t _lock;//互斥鎖
    pthread_cond_t _c_cond;//消費者被喚醒和掛起的條件變數
    pthread_cond_t _p_cond;//生產者被喚醒和掛起的條件變數
}

基本方法的封裝

我對阻塞隊列的一些基本操作進行了封裝，有以下幾個處理動作（可以設置為私有方法）：

  • 判斷隊列為空或為滿
  • 喚醒生產者和喚醒消費者
  • 生產者掛起等待和消費者掛起等待
  • 加鎖和解鎖
private:
  bool IsFull()
  {
    return _q.size() == _capacity;
  }
  bool IsEmpty()
  {
    return _q.empty();
  }
  void ConsumerWait()
  {
    cout << "consumer wait...." << endl;
    pthread_cond_wait(&_c_cond, &_lock);
  }
  void WakeUpConsumer()
  {
    cout << "wake up consumer...." << endl;
    pthread_cond_broadcast(&_c_cond);
  }
  void ProductorWait()
  {
    cout << "productor wait...." << endl;
    pthread_cond_wait(&_p_cond, &_lock);
  }
  void WakeUpProductor()
  {
    cout << "wake up productor...." << endl;
    pthread_cond_broadcast(&_p_cond);
  }
  void LockQueue()
  {
    pthread_mutex_lock(&_lock);
  }
  void UnLockQueue()
  {
    pthread_mutex_unlock(&_lock);
  }

放數據和取數據

  • 生產者進行相關操作前先上鎖，隊列如果為滿就需要掛起等待。隊列不為滿就生成一個數據，然後需要把數據放入阻塞隊列中，解開鎖之後喚醒消費者，喊消費者進行消費。
  • 消費者進行相關操作前先上鎖，隊列如果為空就需要掛起等待。隊列不為空就需要從阻塞隊列中取一個數據，然後解開鎖之後喚醒消費者，喊生產者進行生產。
void ProductData(T data)
{
  LockQueue();
  while (IsFull()){
    // 讓productor掛起等待
    ProductorWait();
  }
  // 放數據
  _q.push(data);
  UnLockQueue();
  // 喚醒consumer
  WakeUpConsumer();
}
void ConsumeData(T& data)
{
  LockQueue();
  while (IsEmpty()){
    // 讓consumer掛起等待
    ConsumerWait();
  }
  // 取數據
  data = _q.front();
  _q.pop();
  UnLockQueue();
  // 喚醒productor
  WakeUpProductor();
}

注意： 在臨界資源判斷喚醒條件是否就緒應該使用while循環檢測，被喚醒的執行緒並不著急立即往下執行，而是再進行一次檢測，判斷當前喚醒條件是否真的就緒了。因為喚醒執行緒的這個函數調用可能會發生失敗，且執行緒可能是在條件不滿足的時候被喚醒，發生誤判被偽喚醒。

封裝一個任務

我們可以實現一個任務類，生產者把這個任務放進阻塞隊列中，消費者取出並進行處理。其中還有一個run的任務執行方法

class Task
{
public:
  Task(int a = 0, int b = 0)
    :_a(a)
    ,_b(b)
  {}
  int Run()
  {
    return _a + _b;
  }
  int GetA()
  {
    return _a;
  }
  int GetB()
  {
    return _b;
  }
private:
  int _a;
  int _b;
};

單生產者和單消費者模型分析

BlockQueue<Task>* q;// 阻塞隊列

void* Consumer(void* arg)
{
  long id = (long)arg;
  while (1){
    // 消費（取）數據
    Task t(0, 0);
    q->ConsumeData(t);
    cout << "consumer " << id << " consumes a task: " << t.GetA() << " + " << t.GetB() << " = " << t.Run() << endl;
    sleep(1);// 後面可注釋，調整速度
  }
}
void* Productor(void* arg)
{
  long id = (long)arg;
  while (1){
    // 生產（放）數據
    int x = rand()%10 + 1;
    int y = rand()%10 + 1;
    Task t(x, y);
    cout << "productor " << id << " produncs a task: " << x << " + " << y << " = ?" << endl;
    q->ProductData(t);
    sleep(1);// 後面可注釋，調整速度
  }
}
int main()
{
  srand((size_t)time(nullptr));

  // 創建一個交易場所
  q =  new BlockQueue<Task>;
  
  pthread_t p, c;
  pthread_create(&p, nullptr, Productor, (void*)(1));
  pthread_create(&c, nullptr, Consumer, (void*)(1));
  
  pthread_join(p, nullptr);
  pthread_join(c, nullptr);
   
  delete q;
  return 0;
}

程式碼運行的結果分三種情況分析：

1.消費者和生產者以相同的速度執行

可以看出的是，生產者生成完一個任務，消費者就處理了一個任務。接著生產者繼續生產，消費者跟著處理，二者步調一致，一直並行的狀態

2.生產者快，消費者慢

生產者生產速度快，導致一上來生產者就把隊列塞滿了任務，接著喚醒消費者來消費，然後掛起等待。緊接著消費者處理完一個任務就喚醒生產者來生產，生產者生產了一個任務就喊消費者來消費，然後繼續掛起。可以看出的是，在這種情況下，隊列長時間是滿的，生產者大部分時間是掛起等待的。生產者和消費者開始一小部分時間內步調是不一致的，生產者生產完，消費者才消費是串列的，但是過了一會，二者步調就變得一致了，且速度是隨消費者的

3.生產者慢，消費者快

生產者的速度慢，生產者生產一個任務立馬喚醒消費者，消費者處理完一個數據，發現隊列為空，然後掛起等待，接著生產者繼續生產一個任務，然後喚醒消費者。可以看出的是，隊列大部分時間是為空的，消費者大部分時間是處於掛起等待的，二者步調一直是一致的，且執行速度是隨生產者的，也是並行的

多生產者和多消費者模型分析

做到幾點：

  • 生產者之間需要互斥，也就是生產者和生產者之間需要組內競爭一把鎖，消費者也是如此
  • 生產者和消費者之間用互斥量和條件變數做到同步和互斥（上面就做到了）
#define P_COUNT 3
#define C_COUNT 3

BlockQueue<Task>* q;

pthread_mutex_t c_lock;// 消費者的鎖
pthread_mutex_t p_lock;// 生產者的鎖

void* Consumer(void* arg)
{
  long id = (long)arg;

  while (1){
    pthread_mutex_lock(&c_lock);
    // 消費（取）數據
    Task t(0, 0);
    q->ConsumeData(t);
    cout << "consumer " << id << " consumes a task: " << t.GetA() << " + " << t.GetB() << " = " << t.Run() << endl;
    pthread_mutex_unlock(&c_lock);
    sleep(1);
  }
}

void* Productor(void* arg)
{
  long id = (long)arg;
  while (1){
    pthread_mutex_lock(&p_lock);
    // 生產（放）數據
    int x = rand()%10 + 1;
    int y = rand()%10 + 1;
    Task t(x, y);
    cout << "productor " << id << " produncs a task: " << x << " + " << y << " = ?" << endl;
    q->ProductData(t);
    pthread_mutex_unlock(&p_lock);
    sleep(1);
  }
}
int main()
{
  srand((size_t)time(nullptr));
  pthread_mutex_init(&c_lock, nullptr);
  pthread_mutex_init(&p_lock, nullptr);

  // 創建一個交易場所
  q =  new BlockQueue<Task>;

  pthread_t p[P_COUNT];
  pthread_t c[C_COUNT];

  for (long i = 0; i < P_COUNT; ++i)
  {
    pthread_create(p+i, nullptr, Productor, (void*)(i+1));
  }
  for (long i = 0; i < C_COUNT; ++i)
  {
    pthread_create(c+i, nullptr, Consumer, (void*)(i+1));
  }
  
  for (int i = 0; i < C_COUNT; ++i)
  {
    pthread_join(c[i], nullptr);
  }
  for (int i = 0; i < P_COUNT; ++i)
  {
    pthread_join(p[i], nullptr);
  }
  
  pthread_mutex_destroy(&c_lock);
  pthread_mutex_destroy(&p_lock);
  
  delete q;
  return 0;
}

注意：

  • 生產者之間需要一個互斥量，消費者之間也需要一個互斥量
  • 生產者和消費者的個數可以自己調整宏變數

運行結果如下：

POSIX訊號量

介紹

POSIX訊號量： 該訊號量允許進程和執行緒同步對共享資源的訪問。同時也可以用於實現執行緒間同步。

總結幾點：

  • 是什麼？ 訊號量本質是一個計數器，描述臨界資源的有效個數。申請一個資源就對訊號量減1（P操作），釋放一個資源就對訊號量加1（V操作）
  • 為什麼？ 臨界資源可以看成很多份，互相不衝突且高效
  • 怎麼用？ 可以使用訊號量的相關介面，來申請訊號量和釋放訊號量（下面詳細介紹）

相關介面的介紹

下面要介紹的POSIX訊號量相關介面都是在semaphore.h的頭文件中。訊號量是一個類型為sem_t的變數

  • 初始化訊號量
int sem_init(sem_t *sem, int pshared, unsigned int value); 
功能：
	創建一個訊號量並初始化它的值。一個無名訊號量在被使用前必須先初始化
參數：
	sem：訊號量的地址
	pshared：等於0，訊號量在執行緒間共享（常用）。不等於0，訊號量在進程間共享
	value：訊號量的初始值
返回值：
	成功：0
	失敗：-1
	
訊號量的初始值就是可用資源的個數
  • 銷毀訊號量
int sem_destroy(sem_t *sem);
功能：
	刪除sem標識的訊號量
參數：
	sem：信量地址
返回值：
	成功：0
	失敗：-1
  • 訊號量P操作（減一）
int sem_wait(sem_t *sem);
功能：
	將訊號量的值減1。操作前，先檢查訊號量(sem)的值是否為0，若訊號量為0，此函數會阻塞，直到訊號量大於0時才進行減一操作
參數：
	sem:訊號量的地址
返回值：
	成功：0
	失敗：-1

int sem_trywait(sem_t *sem)；
	以非阻塞的方式來對訊號量進行減1操作
	若操作前，訊號量的值等於0，則對訊號量的操作失敗，函數立即返回
int sem_timedwait(sem_t* sem，const struct timespec *abs_timeout)；
	限時嘗試將訊號量的值減1
	abs_timeout：絕對時間
  • 訊號量V操作（加一）
int sem_post(sem_t *sem);
功能：
	將訊號量的值加1並發出訊號喚醒等待執行緒(sem_wait())
參數：
	sem：訊號量的地址。
返回值：
成功：0
  • 獲取訊號量的值
int sem_getvalue(sem_t *sem，int *Sval);
功能:
	獲取sem標識的訊號量的值，保存在sval中
參數:
	sem：訊號量地址
	sval：保存訊號值的地址
返回值：
	成功：0
	失敗：-1

舉個簡單的例子：

初始化訊號量的資源數是1，意思就是系統中只有一份資源，執行緒1如果要申請這份資源，首先要進行p操作，sem_wait()函數，那麼資源數-1，此時資源數是0，說明系統中沒有可用的資源，其他執行緒在進行p操作的時候就會阻塞，執行緒1執行完畢，執行v操作，資源數+1，系統中的資源被釋放出來，執行緒2此時解除阻塞狀態，步驟和執行緒1一樣。

程式碼示例：

#include <iostream>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
using namespace std;
sem_t sem;
void* run1(void* arg)
{
  while (1){
    sem_wait(&sem);
    cout << "run1 is running..." << endl;
    sem_post(&sem);
    sleep(1);
  }
}
void* run2(void* arg)
{
  while (1){
    sem_wait(&sem);
    cout << "run2 is running..." << endl;
    sem_post(&sem);
    sleep(1);
  }
}
int main()
{
  sem_init(&sem, 0, 1);
  pthread_t t1, t2;
  pthread_create(&t1, nullptr, run1, nullptr);
  pthread_create(&t2, nullptr, run2, nullptr);

  sem_destroy(&sem);
  pthread_join(t1, nullptr);
  pthread_join(t2, nullptr);
  return 0;
}

運行結果如下：

注意：當訊號量的初始值為1的時候，就相當於一把互斥鎖

基於環形隊列的生產消費模型

環形隊列介紹

環形隊列： 環形隊列和普通隊列的區別就是，這種隊列是一種環形的結構，有一個頭指針和一個尾指針維護環中一小段隊列。

環形結構起始狀態和結束狀態都是一樣的，不好判斷為空或者為滿，所以可以通過加計數器或者標記位來判斷滿或者空。另外也可以預留一個空的位置，作為滿的狀態。

這就類似於數據結構中學的循環隊列類似。

因為訊號量就是一個計數器，所以我們可以通過訊號量來實現多執行緒間的同步過程。

實現

一個交易場所： 循環隊列
兩個角色：

  • 生產者：需要申請空間資源（P操作），然後釋放數據資源（V操作）
  • 消費者：需要申請數據資源（P操作），然後釋放空間資源（V操作）
  • 三種關係： 生產者與生產者（互斥）、生產者與消費者（同步（主要）和互斥）和消費者與消費者（互斥）

幾個變數成員：

  • 隊列：數組模擬
  • 容量：由用戶給定
  • 空間資源訊號量：隊列的容量大小
  • 數據資源訊號量：開始為0
  • 生產者的下標位置：開始為0
  • 消費者的下標位置：開始為0

程式碼示例：

#include<iostream>
#include<string.h>
#include<vector>
#include<semaphore.h>
#include<pthread.h>
#include<unistd.h>
using namespace std;
//生產者：需要申請空間資源（P操作），然後釋放數據資源（V操作）
//消費者：需要申請空間資源（P操作），然後釋放空間資源（V操作）
template<class T>
class RingQueue
{
public:
    RingQueue(int capacity = 5):_capacity(capacity),_rq(capacity),_c_index(0),_p_index(0)
    {
      //初始化空間資源訊號量，容量為5
      sem_init(&_blank_sem,0,_capacity);
      sem_init(&_data_sem,0,0);
    }
    ~RingQueue()
    {
      sem_destroy(&_blank_sem);
      sem_destroy(&_data_sem);
    }
private:
    void P(sem_t& sem)
{
      sem_wait(&sem);
    }
    void V(sem_t& sem)
    {
      sem_post(&sem);
    }
public:
  void GetData(T& data)
  {
    // consumer申請數據資源
    P(_data_sem);
    data = _rq[_c_index];
    _c_index = (_c_index + 1) % _capacity;
    // consumer釋放格子資源
    V(_blank_sem);
  }
  void PutData(const T& data)
  {
    // productor申請格子資源
    P(_blank_sem);
    _rq[_p_index] = data;
    _p_index = (_p_index + 1) % _capacity;
    // productor釋放數據資源
V(_data_sem);
  }

private:
    //vector容器模擬隊列
    vector<T> _rq;
    //隊列的容量
    size_t _capacity;
    //空間資源訊號量
    sem_t _blank_sem;
    //數據資源訊號量
    sem_t _data_sem;
    //消費者的下標位置：開始為0
    int _c_index;
    //生產者的下標位置：開始為0
    int _p_index;
};
RingQueue<int> *q ;
void* Consumer(void* arg)
  {
    long id = (long)arg;
    while (1){
      // 消費（取）數據
      int x;
      q->GetData(x);
      cout << "consumer " << id << " consumes a data: " << x << endl;
      sleep(1);// 後面可注釋，調整速度
   }
}
 
void* Productor(void* arg)
{
    long id = (long)arg;
    while (1){
      // 生產（放）數據
      int x = rand()%10 + 1;
      q->PutData(x);
      cout << "productor " << id << " produncs a data: " << x<< endl;
      sleep(1);// 後面可注釋，調整速度
    }
  }
int main()
{
	// 創建一個交易場所
    q = new RingQueue<int>();
    srand((size_t)time(nullptr));
  
    pthread_t p, c;
    pthread_create(&p, nullptr, Productor, (void*)(1));
    pthread_create(&c, nullptr, Consumer, (void*)(1));
 
    pthread_join(p, nullptr);
    pthread_join(c, nullptr);
 
    delete q;
    return EXIT_SUCCESS;
}

注意：生產者生成數據前需要申請空間資源訊號量（P(_blank_sem)），申請不成功就掛起等待，等待訊號量來了繼續獲得訊號量，然後釋放數據資源訊號量（V(_data_sem)）
消費者消費數據前需要申請數據資源訊號量（P(_data_sem)），申請不成功就掛起等待，等待訊號量來了繼續獲得訊號量，然後釋放空間資源訊號量（V(_blank_sem)）

運行結果如下：

  • 生產者和消費者執行速度一致

生產者生產完一個數據，然後消費者就消費了，二者步調一致，並發執行。

  • 生產者快，消費者慢

生產者速度快，一下字就把隊列塞滿了數據（開始時二者步調不一致），接著生產者如果再去申請空間訊號量，此時已經申請不到了，只能掛起等待，消費者消費數據是否空間訊號量，這是生產者才可以繼續生產，可以看出，在後面大部分時間，二者步調恢復一致了，且速度隨消費者。

  • 生產者慢，消費者快

生產生產者生產完一個數據，數據訊號量加1，空間訊號量減1，然後消費者里馬消費了一個數據，數據訊號量減1，空間訊號量加1，此時數據訊號量為0，消費者再去申請數據訊號量，申請不到就掛起等待，只能等生產者在去生產釋放空間訊號量，然後消費者才可以申請到。可以看出的是，隊列長時間是空的，二者步調一致，速度隨生產者。

Tags:

VirMach 便宜 VPS