設計目標
- 預創建固定數量線程，避免頻繁創建/銷毀開銷。
- 任務隊列線程安全，支持動態提交。
- 工作線程空閒/忙碌狀態跟蹤，優雅關閉。
核心數據結構
// 抽象任務基類class CTask {protected: std::string m_strTaskName; void* m_ptrData; // 業務數據指針public: CTask() : m_ptrData(nullptr) {} explicit CTask(const std::string& name) : m_strTaskName(name), m_ptrData(nullptr) {} virtual ~CTask() {} virtual int Run() = 0; // 純虛函數，子類實現業務邏輯 void SetData(void* data) { m_ptrData = data; } const std::string& GetName() const { return m_strTaskName; }};
線程池類
class CThreadPool {private: std::vector m_vecTaskList; // 任務隊列 std::vector m_vecIdleThread; // 空閒線程 ID std::vector m_vecBusyThread; // 忙碌線程 ID pthread_mutex_t m_mutex; // 隊列/狀態鎖 pthread_cond_t m_cond; // 條件變量: 有任務/線程可用 int m_iThreadNum; // 線程池大小 bool m_bShutdown; // 關閉標誌 // 靜態線程入口函數 static void* ThreadFunc(void* arg); // 狀態遷移輔助 int MoveToIdle(pthread_t tid); int MoveToBusy(pthread_t tid); int CreateThreads();public: explicit CThreadPool(int threadNum); ~CThreadPool(); int AddTask(CTask* task); // 提交任務, 喚醒一個空閒線程 int StopAll(); // 停止所有線程, 清理資源};
關鍵實現細節
線程啟動
int CThreadPool::CreateThreads() { for (int i = 0; i < m_iThreadNum; ++i) { pthread_t tid; if (pthread_create(&tid, nullptr, ThreadFunc, this) != 0) return -1; m_vecIdleThread.push_back(tid); } return 0;}
工作線程主循環
void* CThreadPool::ThreadFunc(void* arg) { CThreadPool* pool = static_cast(arg); while (true) { CTask* task = nullptr; // 1. 獲取任務 (加鎖等待) pthread_mutex_lock(&pool->m_mutex); while (pool->m_vecTaskList.empty() && !pool->m_bShutdown) { pthread_cond_wait(&pool->m_cond, &pool->m_mutex); } if (pool->m_bShutdown) { pthread_mutex_unlock(&pool->m_mutex); break; } task = pool->m_vecTaskList.front(); pool->m_vecTaskList.erase(pool->m_vecTaskList.begin()); pool->MoveToBusy(pthread_self()); pthread_mutex_unlock(&pool->m_mutex); // 2. 執行任務 (無鎖, 並發) if (task) { task->Run(); delete task; // 池負責釋放任務對象 } // 3. 歸還空閒 pthread_mutex_lock(&pool->m_mutex); pool->MoveToIdle(pthread_self()); pthread_mutex_unlock(&pool->m_mutex); } return nullptr;}
提交任務
int CThreadPool::AddTask(CTask* task) { if (!task) return -1; pthread_mutex_lock(&m_mutex); if (m_bShutdown) { pthread_mutex_unlock(&m_mutex); return -1; } m_vecTaskList.push_back(task); pthread_cond_signal(&m_cond); // 喚醒一個等待線程 pthread_mutex_unlock(&m_mutex); return 0;}
優雅關閉
int CThreadPool::StopAll() { pthread_mutex_lock(&m_mutex); m_bShutdown = true; pthread_cond_broadcast(&m_cond); // 喚醒所有等待線程 pthread_mutex_unlock(&m_mutex); for (pthread_t tid : m_vecIdleThread) pthread_join(tid, nullptr); for (pthread_t tid : m_vecBusyThread) pthread_join(tid, nullptr); m_vecIdleThread.clear(); m_vecBusyThread.clear(); // 清理殘留任務 for (auto* t : m_vecTaskList) delete t; m_vecTaskList.clear(); return 0;}
使用示例
class MyTask : public CTask {public: int Run() override { int* data = static_cast(m_ptrData); std::cout << "Processing " << *data << " on thread " << pthread_self() << std::endl; return 0; }};int main() { CThreadPool pool(4); for (int i = 0; i < 10; ++i) { int* p = new int(i); MyTask* task = new MyTask(); task->SetData(p); pool.AddTask(task); } sleep(2); // 等待完成 pool.StopAll(); return 0;}
現代 C++ 改進建議 (C++11/17)
- 用
std::thread、std::mutex、std::condition_variable 替代 pthread。 - 任務類型擦除：
std::function + std::packaged_task 支持任意可調用對象與 future 返回值。 - 智能指針管理任務生命週期：
std::unique_ptr 或 std::shared_ptr。 - 工作盜演算法 (
std::deque + 雙端鎖) 提升多核負載均衡。 - 參考開源成熟庫：progschj/ThreadPool (單頭文件)、bshoshany/thread-pool (功能豐富)。
