前言
實現消息隊列的關鍵因素是考量不同線程訪問消息隊列的同步問題。本實現涉及到幾個知識點
std::lock_guard 介紹
std::lock_gurad 是 C++11 中定義的模板類。定義如下：
template <class Mutex> class lock_guard;
lock_guard 對象通常用于管理某個鎖(Lock)對象，因此與 Mutex RAII 相關，方便線程對互斥量上鎖，即在某個 lock_guard 對象的聲明周期內，它所管理的鎖對象會一直保持上鎖狀態；而 lock_guard 的生命周期結束之后，它所管理的鎖對象會被解鎖(注：類似 shared_ptr 等智能指針管理動態分配的內存資源 )。
模板參數 Mutex 代表互斥量類型，例如 std::mutex 類型，它應該是一個基本的 BasicLockable 類型，標準庫中定義幾種基本的 BasicLockable 類型，分別 std::mutex, std::recursive_mutex, std::timed_mutex，std::recursive_timed_mutex 以及 std::unique_lock
std::unique_lock 介紹
lock_guard 最大的缺點也是簡單，沒有給程序員提供足夠的靈活度，因此，C++11 標準中定義了另外一個與 Mutex RAII 相關類 unique_lock，該類與 lock_guard 類相似，也很方便線程對互斥量上鎖，但它提供了更好的上鎖和解鎖控制。
顧名思義，unique_lock 對象以獨占所有權的方式（ unique owership）管理 mutex 對象的上鎖和解鎖操作，所謂獨占所有權，就是沒有其他的 unique_lock 對象同時擁有某個 mutex 對象的所有權。
新創建的 unique_lock 對象管理 Mutex 對象 m，并嘗試調用 m.lock() 對 Mutex 對象進行上鎖，如果此時另外某個 unique_lock 對象已經管理了該 Mutex 對象 m，則當前線程將會被阻塞。
std::condition介紹
當 std::condition_variable 對象的某個 wait 函數被調用的時候，它使用 std::unique_lock(通過 std::mutex) 來鎖住當前線程。當前線程會一直被阻塞，直到另外一個線程在相同的 std::condition_variable 對象上調用了 notification 函數來喚醒當前線程。
std::condition_variable 提供了兩種 wait() 函數。當前線程調用 wait() 后將被阻塞(此時當前線程應該獲得了鎖（mutex），不妨設獲得鎖 lck)，直到另外某個線程調用 notify_* 喚醒了當前線程。
在線程被阻塞時，該函數會自動調用 lck.unlock() 釋放鎖，使得其他被阻塞在鎖競爭上的線程得以繼續執行。另外，一旦當前線程獲得通知(notified，通常是另外某個線程調用 notify_* 喚醒了當前線程)，wait() 函數也是自動調用 lck.lock()，使得 lck 的狀態和 wait 函數被調用時相同。
在第二種情況下（即設置了 Predicate），只有當 pred 條件為 false 時調用 wait() 才會阻塞當前線程，并且在收到其他線程的通知后只有當 pred 為 true 時才會被解除阻塞。因此第二種情況類似以下代碼：
while (!pred()) wait(lck);
std::function介紹
使用std::function可以將普通函數，lambda表達式和函數對象類統一起來。它們并不是相同的類型，然而通過function模板類，可以轉化為相同類型的對象（function對象），從而放入一個vector或其他容器里，方便回調。
代碼實現：
#pragma once
#ifndef MESSAGE_QUEUE_H
#define MESSAGE_QUEUE_H
#include <queue>
#include <mutex>
#include <condition_variable>
template<class Type>
class CMessageQueue
{
public:
CMessageQueue& operator = (const CMessageQueue&) = delete;
CMessageQueue(const CMessageQueue& mq) = delete;
CMessageQueue() :_queue(), _mutex(), _condition(){}
virtual ~CMessageQueue(){}
void Push(Type msg){
std::lock_guard <std::mutex> lock(_mutex);
_queue.push(msg);
//當使用阻塞模式從消息隊列中獲取消息時，由condition在新消息到達時提醒等待線程
_condition.notify_one();
}
//blocked定義訪問方式是同步阻塞或者非阻塞模式
bool Pop(Type& msg, bool isBlocked = true){
if (isBlocked)
{
std::unique_lock <std::mutex> lock(_mutex);
while (_queue.empty())
{
_condition.wait(lock);
}
//注意這一段必須放在if語句中，因為lock的生命域僅僅在if大括號內
msg = std::move(_queue.front());
_queue.pop();
return true;
}
else
{
std::lock_guard<std::mutex> lock(_mutex);
if (_queue.empty())
return false;
msg = std::move(_queue.front());
_queue.pop();
return true;
}
}
int32_t Size(){
std::lock_guard<std::mutex> lock(_mutex);
return _queue.size();
}
bool Empty(){
std::lock_guard<std::mutex> lock(_mutex);
return _queue.empty();
}
private:
std::queue<Type> _queue;//存儲消息的隊列
mutable std::mutex _mutex;//同步鎖
std::condition_variable _condition;//實現同步式獲取消息
};
#endif//MESSAGE_QUEUE_H
線程池可以直接在構造函數中構造線程，并傳入回調函數，也可以寫一個Run函數顯示調用。這里我們選擇了第二種，對比：
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
#include <functional>
#include <vector>
#include <thread>
#include "MessageQueue.h"
#define MIN_THREADS 1
template<class Type>
class CThreadPool
{
CThreadPool& operator = (const CThreadPool&) = delete;
CThreadPool(const CThreadPool& other) = delete;
public:
CThreadPool(int32_t threads,
std::function<void(Type& record, CThreadPool<Type>* pSub)> handler);
virtual ~CThreadPool();
void Run();
virtual void PreHandler(){}
virtual void AfterHandler(){}
void Submit(Type record);
private:
bool _shutdown;
int32_t _threads;
std::function<void(Type& record, CThreadPool<Type>* pSub)> _handler;
std::vector<std::thread> _workers;
CMessageQueue<Type> _tasks;
};
template<class Type>
CThreadPool<Type>::CThreadPool(int32_t threads,
std::function<void(Type& record, CThreadPool<Type>* pSub)> handler)
:_shutdown(false),
_threads(threads),
_handler(handler),
_workers(),
_tasks()
{
//第一種實現方案，注意這里的虛函數調用不正確
/*if (_threads < MIN_THREADS)
_threads = MIN_THREADS;
for (int32_t i = 0; i < _threads; i++)
{
_workers.emplace_back(
[this]{
PreHandler();
while (!_shutdown){
Type record;
_tasks.Pop(record, true);
_handler(record, this);
}
AfterHandler();
}
);
}*/
}
//第二種實現方案
template<class Type>
void CThreadPool<Type>::Run()
{
if (_threads < MIN_THREADS)
_threads = MIN_THREADS;
for (int32_t i = 0; i < _threads; i++)
{
_workers.emplace_back(
[this]{
PreHandler();
while (!_shutdown){
Type record;
_tasks.Pop(record, true);
_handler(record, this);
}
AfterHandler();
}
);
}
}
template<class Type>
CThreadPool<Type>::~CThreadPool()
{
for (std::thread& worker : _workers)
worker.join();
}
template<class Type>
void CThreadPool<Type>::Submit(Type record)
{
_tasks.Push(record);
}
#endif // !THREAD_POOL_H
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，謝謝大家對億速云的支持。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
