在高併發場景下，若採用「一請求一線程」的傳統模式，頻繁創建銷毀線程會耗盡系統資源，上下文切換開銷也極大。線程池技術通過預創建固定數量線程、任務完成後重用線程，完美解決了這一問題。本文深入剖析半同步半異步線程池的三層架構，並提供基於 C++11 標準庫的完整實現代碼。
半同步半異步模式將系統劃分為三個邏輯層次，實現任務接收與處理解耦：
負責接收上層並發任務請求。請求到達時不直接處理，而是封裝為任務對象放入同步隊列，實現「快速接收、緩衝排隊」。
核心是一個線程安全的有界阻塞隊列。生產者（同步層）在隊列滿時阻塞等待；消費者（異步層）在隊列空時阻塞等待。該層起到流量削峰填谷、解耦生產消費速率的關鍵作用。
由多個工作線程組成，並發從隊列取出任務執行。線程數通常設置為 CPU 核心數，最大化 CPU 利用率，避免過度上下文切換。
同步隊列是線程池的心臟，基於 std::mutex、std::condition_variable 實現生產消費模型：
m_notEmpty 通知消費者有任務，m_notFull 通知生產者有空位m_needStop 配合 notify_all 喚醒所有等待線程退出Take(list) 一次移動整個隊列，減少鎖競爭，適合任務密集型場景ThreadPool 類封裼了線程組管理與任務調度：
hardware_concurrency()，自動適配硬件RunInThread 持續批量取任務、逐個執行，m_running 原子標誌控制生命週期AddTask 完美轉發任務到同步隊列std::call_once 確保 StopThreadGroup 僅執行一次，join 等待所有線程退出#include <list>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <iostream>
#include <memory>
#include <functional>
#include <atomic>
using namespace std;
/******************************** 同步隊列 *******************************/
template <typename T>
class SyncQueue {
public:
SyncQueue(int maxSize) : m_maxSize(maxSize), m_needStop(false) {}
// 添加任務（左值/右值完美轉發）
void Put(const T& x) { Add(x); }
void Put(T&& x) { Add(std::forward<T>(x)); }
// 批量取走所有任務
void Take(std::list<T>& list) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notEmpty.wait(locker, [this] { return m_needStop || NotEmpty(); });
if (m_needStop) return;
list = std::move(m_queue);
m_notFull.notify_one();
}
// 取單個任務
void Take(T& t) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notEmpty.wait(locker, [this] { return m_needStop || NotEmpty(); });
if (m_needStop) return;
t = m_queue.front();
m_queue.pop_front();
m_notFull.notify_one();
}
void Stop() {
{
std::lock_guard<std::mutex> locker(m_mutex);
m_needStop = true;
}
m_notFull.notify_all();
m_notEmpty.notify_all();
}
bool Empty() { std::lock_guard<std::mutex> locker(m_mutex); return m_queue.empty(); }
bool Full() { std::lock_guard<std::mutex> locker(m_mutex); return m_queue.size() == m_maxSize; }
size_t Size() { std::lock_guard<std::mutex> locker(m_mutex); return m_queue.size(); }
private:
bool NotFull() const {
bool full = (m_queue.size() >= m_maxSize);
if (full) cout << "the queue is full, need wait..." << endl;
return !full;
}
bool NotEmpty() const {
bool empty = m_queue.empty();
if (empty) cout << "the queue is empty, need wait..., thread ID: " << this_thread::get_id() << endl;
return !empty;
}
template <typename F>
void Add(F&& x) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notFull.wait(locker, [this] { return m_needStop || NotFull(); });
if (m_needStop) return;
m_queue.push_back(std::forward<F>(x));
m_notEmpty.notify_one();
}
private:
std::list<T> m_queue;
std::mutex m_mutex;
std::condition_variable m_notEmpty;
std::condition_variable m_notFull;
int m_maxSize;
bool m_needStop;
};
/************************** 線程池 *********************************/
const int MaxTaskCount = 100;
class ThreadPool {
public:
using Task = std::function<void()>;
ThreadPool(int numThreads = std::thread::hardware_concurrency()) : m_queue(MaxTaskCount) {
cout << "numThreads: " << numThreads << endl;
Start(numThreads);
}
~ThreadPool() { Stop(); }
void Stop() {
std::call_once(m_flag, [this] { StopThreadGroup(); });
}
void AddTask(Task&& task) { m_queue.Put(std::forward<Task>(task)); }
void AddTask(const Task& task) { m_queue.Put(task); }
private:
void Start(int numThreads) {
m_running = true;
for (int i = 0; i < numThreads; ++i) {
m_threadgroup.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
}
}
void RunInThread() {
while (m_running) {
std::list<Task> list;
m_queue.Take(list);
for (auto& task : list) {
if (!m_running) return;
task();
}
}
}
void StopThreadGroup() {
m_queue.Stop();
m_running = false;
for (auto& thread : m_threadgroup) {
if (thread) thread->join();
}
m_threadgroup.clear();
}
private:
std::list<std::shared_ptr<std::thread>> m_threadgroup;
SyncQueue<Task> m_queue;
std::atomic_bool m_running{false};
std::once_flag m_flag;
};
/************************** 測試主程序 *******************************/
int main() {
ThreadPool pool;
std::thread thd1([&pool] {
for (int i = 0; i < 10; ++i) {
auto thdId = this_thread::get_id();
pool.AddTask([thdId] { cout << "1.thread id: " << thdId << endl; });
}
});
std::thread thd2([&pool] {
for (int i = 0; i < 10; ++i) {
auto thdId = this_thread::get_id();
pool.AddTask([thdId] { cout << "2.thread id: " << thdId << endl; });
}
});
this_thread::sleep_for(std::chrono::seconds(2));
getchar();
pool.Stop();
thd1.join();
thd2.join();
return 0;
}
對象池適用於創建開銷大的對象（如數據庫連接、線程對象）。核心思想：預創建對象存入容器，使用時取出，歸還時放回，避免重複構造析構。
關鍵實現要點：
NonCopyable 禁止拷貝multimap<string, shared_ptr<T>> 按構造參數類型分類管理shared_ptr 刪除器：析構時不釋放內存，而是歸還對象池typeid(Constructor).name() 作為鍵區分不同構造簽名#include <string>
#include <functional>
#include <memory>
#include <map>
using namespace std;
class NonCopyable {
protected:
NonCopyable() = default;
~NonCopyable() = default;
NonCopyable(const NonCopyable&) = delete;
NonCopyable& operator=(const NonCopyable&) = delete;
};
const int MaxObjectNum = 10;
template <typename T>
class ObjectPool : NonCopyable {
template <typename... Args>
using Constructor = function<shared_ptr<T>(Args...)>;
private:
multimap<string, shared_ptr<T>> m_object_map;
public:
template <typename... Args>
void Init(size_t num, Args... args) {
if (num <= 0 || num > MaxObjectNum) throw logic_error("Object num out of range");
auto constructName = typeid(Constructor<Args...>).name();
for (size_t i = 0; i < num; ++i) {
m_object_map.emplace(constructName,
shared_ptr<T>(new T(std::forward<Args>(args)...),
[this, constructName](T* p) {
cout << "dis: " << constructName << endl;
m_object_map.emplace(std::move(constructName), shared_ptr<T>(p));
}));
}
}
template <typename... Args>
shared_ptr<T> Get() {
string constructName = typeid(Constructor<Args...>).name();
auto range = m_object_map.equal_range(constructName);
for (auto it = range.first; it != range.second; ++it) {
auto ptr = it->second;
m_object_map.erase(it);
return ptr;
}
return nullptr;
}
};
