前言
實現消息隊列的關鍵因素是考量不同線程訪問消息隊列的同步問題。本實現涉及到幾個知識點
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 1template<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總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，謝謝大家對武林網的支持。
新聞熱點
疑難解答
圖片精選
