利用C++如何實現一個阻塞佇列詳解

阿新 • • 發佈：2020-10-07

阻塞佇列是多執行緒中常用的資料結構，對於實現多執行緒之間的資料交換、同步等有很大作用。

阻塞佇列常用於生產者和消費者的場景，生產者是向佇列裡新增元素的執行緒，消費者是從佇列裡取元素的執行緒。簡而言之，阻塞佇列是生產者用來存放元素、消費者獲取元素的容器。

考慮下，這樣一個多執行緒模型，程式有一個主執行緒 master 和一些 worker 執行緒，master 執行緒負責接收到資料，給 worker 執行緒分配資料，worker 執行緒取得一個任務後便可以開始工作，如果沒有任務便阻塞住，節約 cpu 資源。

master 執行緒（生產者）：負責往阻塞佇列中塞入資料，並喚醒正在阻塞的 worker 執行緒。

workder 執行緒（消費者）：負責從阻塞佇列中取資料，如果沒有資料便阻塞，直到被 master 執行緒喚醒。

那麼怎樣的資料結構比較適合做這樣的喚醒呢？顯而易見，是條件變數，在 c++ 11 中，stl 已經引入了執行緒支援庫。

C++11 中條件變數

條件變數一般與一個互斥量同時使用，使用時需要先給互斥量上鎖，然後條件變數會檢測是否滿足條件，如果不滿足條件便會暫時釋放鎖，然後阻塞執行緒。

c++ 11使用方法主要如下：

#include <mutex>
#include <condition_value>
// 互斥量與條件變數
std::mutex m_mutex;
std::condition_value m_condition;

// 請求訊號的一方
std::unique_lock<std::mutex> lock(mutex);
while(xxx)
{
 // 這裡會先釋放 lock，
 // 如果有訊號喚醒的話，會重新加鎖。
 m_condition.wait(lock);
}

// 傳送訊息進行同步的一方
{
 std::unique_lock<std::mutex> lock(mutex);
 // 喚醒其他正在 wait 的執行緒
 m_condition.notify_all();
}

用 C++11 實現阻塞佇列

我們使用條件變數包裝 STL 中的 queue 就可以實現阻塞佇列功能，如果有興趣，甚至可以自己實現一個效率更高的佇列資料結構。

我們先假設一下阻塞佇列需要如下介面：

push 將一個變數塞入佇列；
take 從佇列中取出一個元素；
size 檢視佇列有多少個元素；

template <typename T>
class BlockingQueue
{
public:
 BlockingQueue();
 void push(T&& value);
 T take();
 size_t size() const;
 
private:
 // 實際使用的資料結構佇列
 std::queue<T> m_data;

 // 條件變數
 std::mutex m_mutex;
 std::condition_variable m_condition;
};

push 一個變數時，我們需要先加鎖，加鎖成功後才可以壓入變數，這是為了執行緒安全。壓入變數後，就可以傳送訊號通知正在阻塞的條件變數。

 void push(T&& value)
 {
  // 往佇列中塞資料前要先加鎖
  std::unique_lock<std::mutex> lock(m_mutex);
  m_data.push(value);
  // 喚醒正在阻塞的條件變數
  m_condition.notify_all();
 }

take 一個變數時，就要有些不一樣：

先加鎖，加鎖成功後，如果佇列不為空，可以直接取資料，不需要 wait；
如果佇列為空呢，則 wait 等待，直到被喚醒；
考慮特殊情況，喚醒後佇列依然是空的……

 T take()
 {
  std::unique_lock<std::mutex> lock(m_mutex);
  while(m_data.empty())
  {
   // 等待
   m_condition.wait(lock);
  }
  assert(!m_data.empty());
  T value(std::move(m_data.front()));
  m_data.pop();

  return value;
 }

總結下，程式碼如下：

#ifndef BLOCKINGQUEUE_H
#define BLOCKINGQUEUE_H

#include <queue>
#include <mutex>
#include <condition_variable>
#include <assert.h>

template <typename T>
class BlockingQueue
{
public:
 BlockingQueue()
  :m_mutex(),m_condition(),m_data()
 {
 }

 // 禁止拷貝構造
 BlockingQueue(BlockingQueue&) = delete;

 ~BlockingQueue()
 {
 }

 void push(T&& value)
 {
  // 往佇列中塞資料前要先加鎖
  std::unique_lock<std::mutex> lock(m_mutex);
  m_data.push(value);
  m_condition.notify_all();
 }

 void push(const T& value)
 {
  std::unique_lock<std::mutex> lock(m_mutex);
  m_data.push(value);
  m_condition.notify_all();
 }

 T take()
 {
  std::unique_lock<std::mutex> lock(m_mutex);
  while(m_data.empty())
  {
   m_condition.wait(lock);
  }
  assert(!m_data.empty());
  T value(std::move(m_data.front()));
  m_data.pop();

  return value;
 }

 size_t size() const
 {
  std::unique_lock<std::mutex> lock(m_mutex);
  return m_data.size();
 }
private:
 // 實際使用的資料結構佇列
 std::queue<T> m_data;

 // 條件變數的鎖
 std::mutex m_mutex;
 std::condition_variable m_condition;
};
#endif // BLOCKINGQUEUE_H

實驗程式碼

我們寫個簡單的程式實驗一下，下面程式有一個 master 執行緒，5個 worker 執行緒，master執行緒生成一個隨機數，求 0-隨機數的和。

#include <iostream>
#include <thread>
#include <mutex>
#include <random>

#include <windows.h>

#include <blockingqueue.h>
using namespace std;

int task=100;
BlockingQueue<int> blockingqueue;
std::mutex mutex_cout;

void worker()
{
 int value;
 thread::id this_id = this_thread::get_id();
 while(true)
 {
  value = blockingqueue.take();
  uint64_t sum = 0;
  for(int i = 0; i < value; i++)
  {
   sum += i;
  }

  // 模擬耗時操作
  Sleep(100);

  std::lock_guard<mutex> lock(mutex_cout);
  std::cout << "workder: " << this_id << " "
     << __FUNCTION__
     << " line: " << __LINE__
     << " sum: " << sum
     << std::endl;
 }
}

void master()
{
 srand(time(nullptr));
 for(int i = 0; i < task; i++)
 {
  blockingqueue.push(rand()%10000);
  printf("%s %d %i\n",__FUNCTION__,__LINE__,i);
  Sleep(20);
 }
}

int main()
{
 thread th_master(master);
 std::vector<thread> th_workers;
 for(int i =0; i < 5; i++)
 {
  th_workers.emplace_back(thread(worker));
 }

 th_master.join();
 return 0;
}

從輸出結果可以看出，master 執行緒將任務分配給了正在空閒的 worker 執行緒，具體是哪個執行緒就看作業系統的隨機排程了。

master 46 5
worker: 3 worker line: 34 sum: 20998440
master 46 6
worker: 7 worker line: 34 sum: 3308878
master 46 7
worker: 4 worker line: 34 sum: 34598721
master 46 8
worker: 6 worker line: 34 sum: 1563796
master 46 9
worker: 5 worker line: 34 sum: 27978940

Reference

條件變數

總結

到此這篇關於利用C++如何實現一個阻塞佇列的文章就介紹到這了,更多相關C++實現阻塞佇列內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！