我們知道Java語言對於多執行緒的支援十分豐富，JDK本身提供了很多效能優良的庫，包括ThreadPoolExecutor和ScheduleThreadPoolExecutor等。C++11中的STL也提供了std:thread（然而我還沒有看，這裡先佔個坑）還有很多第三方庫的實現。這裡我重複“造輪子”的目的還是為了深入理解C++和Linux執行緒基礎概念，主要以學習的目的。

首先，為什麼要使用執行緒池。因為執行緒的建立、和清理都是需要耗費系統資源的。我們知道Linux中執行緒實際上是由輕量級程序實現的，相對於純理論上的執行緒這個開銷還是有的。假設某個執行緒的建立、執行和銷燬的時間分別為T1、T2、T3，當T1+T3的時間相對於T2不可忽略時，執行緒池的就有必要引入了，尤其是處理數百萬級的高併發處理時。執行緒池提升了多執行緒程式的效能，因為執行緒池裡面的執行緒都是現成的而且能夠重複使用，我們不需要臨時建立大量執行緒，然後在任務結束時又銷燬大量執行緒。一個理想的執行緒池能夠合理地動態調節池內執行緒數量，既不會因為執行緒過少而導致大量任務堆積，也不會因為執行緒過多了而增加額外的系統開銷。

其實執行緒池的原理非常簡單，它就是一個非常典型的生產者消費者同步問題。根據剛才描述的執行緒池的功能，可以看出執行緒池至少有兩個主要動作，一個是主程式不定時地向執行緒池新增任務，另一個是執行緒池裡的執行緒領取任務去執行。且不論任務和執行任務是個什麼概念，但是一個任務肯定只能分配給一個執行緒執行。這樣就可以簡單猜想執行緒池的一種可能的架構了：主程式執行入隊操作，把任務新增到一個佇列裡面；池子裡的多個工作執行緒共同對這個佇列試圖執行出隊操作，這裡要保證同一時刻只有一個執行緒出隊成功，搶奪到這個任務，其他執行緒繼續共同試圖出隊搶奪下一個任務。所以在實現執行緒池之前，我們需要一個佇列。這裡的生產者就是主程式，生產任務（增加任務），消費者就是工作執行緒，消費任務（執行、減少任務）。因為這裡涉及到多個執行緒同時訪問一個佇列的問題，所以我們需要互斥鎖來保護佇列，同時還需要條件變數來處理主執行緒通知任務到達、工作執行緒搶奪任務的問題。

一般來說實現一個執行緒池主要包括以下4個組成部分：

1. 執行緒管理器：用於建立並管理執行緒池。
2. 工作執行緒：執行緒池中實際執行任務的執行緒。在初始化執行緒時會預先建立好固定數目的執行緒在池中，這些初始化的執行緒一般處於空閒狀態。
3. 任務介面：每個任務必須實現的介面。當執行緒池的任務佇列中有可執行任務時，被空間的工作執行緒調去執行（執行緒的閒與忙的狀態是通過互斥量實現的），把任務抽象出來形成一個介面，可以做到執行緒池與具體的任務無關。
4. 任務佇列：用來存放沒有處理的任務。提供一種緩衝機制。實現這種結構有很多方法，常用的有佇列和連結串列結構。

流程圖如下：

ool.h

#ifndef __THREAD_POOL_H
#define
 
 __THREAD_POOL_H

#include <vector>
#include <string>
#include <pthread.h>

using namespace std;

/*執行任務的類：設定任務資料並執行*/
class CTask {
protected:
    string m_strTaskName;   //任務的名稱
    void* m_ptrData;    //要執行的任務的具體資料

public:
    CTask() = default;
    CTask(string &taskName): m_strTaskName(taskName), m_ptrData(NULL) {}
    virtualint Run() = 0;
    void setData(void* data);   //設定任務資料
  
    virtual ~CTask() {}
    
};

/*執行緒池管理類*/
class CThreadPool {
private:
    static vector<CTask*> m_vecTaskList;    //任務列表
    static bool shutdown;   //執行緒退出標誌
    int m_iThreadNum;   //執行緒池中啟動的執行緒數
    pthread_t *pthread_id;
  
    static pthread_mutex_t m_pthreadMutex;  //執行緒同步鎖
    static pthread_cond_t m_pthreadCond;    //執行緒同步條件變數
  
protected:
    staticvoid* ThreadFunc(void *threadData);  //新執行緒的執行緒回撥函式
    staticint MoveToIdle(pthread_t tid);   //執行緒執行結束後，把自己放入空閒執行緒中
    staticint MoveToBusy(pthread_t tid);   //移入到忙碌執行緒中去
    int Create();   //建立執行緒池中的執行緒
  
public:
    CThreadPool(int threadNum);
    int AddTask(CTask *task);   //把任務新增到任務佇列中
    int StopAll();  //使執行緒池中的所有執行緒退出
    int getTaskSize();  //獲取當前任務佇列中的任務數
};

#endif

2 thread_pool.cpp

#include "thread_pool.h"
#include <cstdio>

void CTask::setData(void* data) {
    m_ptrData = data;
}

//靜態成員初始化
vector<CTask*> CThreadPool::m_vecTaskList;
bool CThreadPool::shutdown = false;
pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;

//執行緒管理類建構函式
CThreadPool::CThreadPool(int threadNum) {
    this->m_iThreadNum = threadNum;
    printf("I will create %d threads.\n", threadNum);
    Create();
}

//執行緒回撥函式
void* CThreadPool::ThreadFunc(void* threadData) {
    pthread_t tid = pthread_self();
    while(1)
    {
        pthread_mutex_lock(&m_pthreadMutex);
        //如果佇列為空，等待新任務進入任務佇列
        while(m_vecTaskList.size() == 0 && !shutdown)
            pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);
        
        //關閉執行緒
        if(shutdown)
        {
            pthread_mutex_unlock(&m_pthreadMutex);
            printf("[tid: %lu]\texit\n", pthread_self());
            pthread_exit(NULL);
        }
        
        printf("[tid: %lu]\trun: ", tid);
        vector<CTask*>::iterator iter = m_vecTaskList.begin();
        //取出一個任務並處理之
        CTask* task = *iter;
        if(iter != m_vecTaskList.end())
        {
            task = *iter;
            m_vecTaskList.erase(iter);
        }
        
        pthread_mutex_unlock(&m_pthreadMutex);
        
        task->Run();    //執行任務
        printf("[tid: %lu]\tidle\n", tid);
        
    }
    
    return (void*)0;
}

//往任務佇列裡新增任務併發出線程同步訊號
int CThreadPool::AddTask(CTask *task) { 
    pthread_mutex_lock(&m_pthreadMutex);    
    m_vecTaskList.push_back(task);  
    pthread_mutex_unlock(&m_pthreadMutex);  
    pthread_cond_signal(&m_pthreadCond);    
    
    return 0;
}

//建立執行緒
int CThreadPool::Create() { 
    pthread_id = new pthread_t[m_iThreadNum];
        for(int i = 0; i < m_iThreadNum; i++)
            pthread_create(&pthread_id[i], NULL, ThreadFunc, NULL);
        
    return 0;
}

//停止所有執行緒
int CThreadPool::StopAll() {    
    //避免重複呼叫
    if(shutdown)
        return -1;
    printf("Now I will end all threads!\n\n");
    
    //喚醒所有等待程序，執行緒池也要銷燬了
    shutdown = true;
    pthread_cond_broadcast(&m_pthreadCond);
    
    //清楚殭屍
    for(int i = 0; i < m_iThreadNum; i++)
        pthread_join(pthread_id[i], NULL);
    
    delete[] pthread_id;
    pthread_id = NULL;
    
    //銷燬互斥量和條件變數
    pthread_mutex_destroy(&m_pthreadMutex);
    pthread_cond_destroy(&m_pthreadCond);
    
    return 0;
}

//獲取當前佇列中的任務數
int CThreadPool::getTaskSize() {    
    return m_vecTaskList.size();
}

3 main.cpp

#include "thread_pool.h"
#include <cstdio>
#include <stdlib.h>
#include <unistd.h>

class CMyTask: public CTask {   
public:
    CMyTask() = default;    
    int Run() {
        printf("%s\n", (char*)m_ptrData);
        int x = rand()%4 + 1;
        sleep(x);   
        return 0;
    }
    ~CMyTask() {}
};

int main() {
    CMyTask taskObj;
    char szTmp[] = "hello!";
    taskObj.setData((void*)szTmp);
    CThreadPool threadpool(5);  //執行緒池大小為5
    
    for(int i = 0; i < 10; i++)
        threadpool.AddTask(&taskObj);
    
    while(1) {
        printf("There are still %d tasks need to handle\n", threadpool.getTaskSize());
        //任務佇列已沒有任務了
        if(threadpool.getTaskSize()==0) {
            //清除執行緒池
            if(threadpool.StopAll() == -1) {
                printf("Thread pool clear, exit.\n");
                exit(0);
            }
        }
        sleep(2);
        printf("2 seconds later...\n");
    }   
    return 0;
}

4 Makefile

CC:= g++
TARGET:= threadpool
INCLUDE:= -I./
LIBS:= -lpthread
# C++語言編譯引數  
CXXFLAGS:= -std=c++11 -g -Wall -D_REENTRANT
# C預處理引數
# CPPFLAGS:=
OBJECTS :=thread_pool.o main.o
  
$(TARGET): $(OBJECTS)
    $(CC) -o $(TARGET) $(OBJECTS) $(LIBS)
  
# [email protected]表示所有目標集  
%.o:%.cpp   
    $(CC) -c $(CXXFLAGS) $(INCLUDE) $< -o [email protected]
  
.PHONY : clean
clean:   
    -rm -f $(OBJECTS) $(TARGET)

5 輸出結果

I will create 5 threads.
There are still 10 tasks need to handle
[tid: 140056759576320]  run: hello!
[tid: 140056751183616]  run: hello!
[tid: 140056742790912]  run: hello!
[tid: 140056734398208]  run: hello!
[tid: 140056767969024]  run: hello!
2 seconds later...
There are still 5 tasks need to handle
[tid: 140056742790912]  idle
[tid: 140056742790912]  run: hello!
[tid: 140056767969024]  idle
[tid: 140056767969024]  run: hello!
[tid: 140056751183616]  idle
[tid: 140056751183616]  run: hello!
[tid: 140056759576320]  idle
[tid: 140056759576320]  run: hello!
[tid: 140056751183616]  idle
[tid: 140056751183616]  run: hello!
[tid: 140056734398208]  idle
2 seconds later...
There are still 0 tasks need to handle
Now I will end all threads!
2 seconds later...
[tid: 140056734398208]  exit
[tid: 140056767969024]  idle
[tid: 140056767969024]  exit
[tid: 140056759576320]  idle
[tid: 140056759576320]  exit
[tid: 140056751183616]  idle
[tid: 140056751183616]  exit
[tid: 140056742790912]  idle
[tid: 140056742790912]  exit
2 seconds later...
There are still 0 tasks need to handle
Thread pool clear, exit.

擴充套件資料：