在C++11標準之前，多線程編程只能使用pthread_xxx開頭的一組POSIX標準的接口。從C++11標準開始，多線程相關接口封裝在了C++的std命名空間里。

在Linux中，即便我們用std命名空間中的接口，也需要使用-lpthread鏈接pthread庫，不難猜出，C++多線程的底層依然還是POSIX標準的接口。你可能會有疑問，既然底層還是使用pthread庫，為什么不直接用pthread，而要繞一大圈再封裝一層呢？

在我看來，除了統一C++編程接口之外，C++11標準更多的是在語言層面做了很多優化，規避了原來C語言中的很多陷阱，比如C++11中的lock_guard、future、promise等技術，將原來C語言中語法上容易犯錯的地方進行了簡化，簡單來說就是將原來依賴人的地方交給了編譯器（很多時候機器比人更可靠）。比如在C++11標準之前，我們使用mutex是像下面這樣的：

pthread_mutex_lock(mutex)
....
if (condition) {
  ...
} elser if {
  ...
} else {
  ...
}
...
pthread_mutex_unlock(mutex)

lock_gruard locker(mutex)
...
if (condition) {
  ...
} elser if {
  ...
} else {
  ...
}
...

C++編譯器會自動為我們插入釋放鎖的代碼，這樣我們就不用時刻擔心鎖有沒有正確釋放了。我個人的感覺是，從C++11標準開始，C++變得不那么可怕了，甚至很多時候覺得C++變得好用了。

這篇文章我們試著使用C++11標準的新特性來實現一個通用的線程池。首先，我們來看一下C++11標準中線程是如何創建的，先看代碼：

#include 
#include 


using namespace std;
void threadFunc(int &a) {
    a += 10;
    std::cout << "this is thread fun!" << std::endl;
}


int main() {
    int x = 10;
    std::thread t1(threadFunc, std::ref(x));
    t1.join();


    std::cout << "x=" << x << std::endl;
}

使用std::thread創建一個t1線程對象，傳入線程主函數，將x以引用的方式傳入線程主函數，接著調用join方法等待主函數threadFunc執行完成。上面x最終的結果將是20。

整個流程和使用pthread_create其實區別不大，只是C++11將線程的創建封裝成了一個類，使得我們可以用面向對象編程的方式來創建多線程。

我們可以思考一下，要實現一個線程池，應該怎么做？這個問題我們可以倒過來想一下，線程池應該怎么使用。比如，對于一個web服務器來講，為了能利用多核的優勢，我們可以在主線程里接收到請求之后，將對這個請求的具體操作交給worker線程，比如像下面這樣：

這個流程我們在C語言網絡編程那個系列文章中有非常詳細的說明，如果你get不到這個例子是什么意思，建議你去看一下那個系列的文章（C語言網絡編程系列合集）

主線程將請求交給Worker線程這個好理解，但是你有沒有想過，它怎么就能交給Worker線程呢？線程創建完之后，對應的主函數就已經運行起來了，對應accept出來的套接字要怎么傳遞給一個正在運行的函數呢？

為了說明這個問題，我們先來看一段Golang的代碼，暫時看不懂沒關系，后面我會解釋

func TestWorker(t *testing.T) {
  msg := make(chan int, 1)
  notify := make(chan struct{}, 0)
  poolSize := 10


  buildWorkers(poolSize, msg, notify)


  // 模擬任務
  for i := 0; i < poolSize; i++ {
    msg <- i
  }


  for i := 0; i < poolSize; i++ {
    <-notify
  }
}


func buildWorkers(poolSize int, msg <-chan int, notify chan<- struct{}) {
  for i := 0; i < poolSize; i++ {
    go func(i int) {
      if ret := recover(); ret != nil {
        log.Printf("panic: %v", ret)
      }


      log.Println("worker-id:", i)


      for {
        select {
        case m := <-msg:
          fmt.Println("m:", m)
          notify <- struct{}{}
        }
      }
    }(i)
  }
}

buildWorkers方法創建了10個協程，每個協程里面都有一個for循環，一開始每個for循環都阻塞在select語句中的case m := ←msg，如果之前沒有接觸過Go語言（這里的select你可以簡單和Linux中的select技術類比）。另外還有兩個通道，一個是msg和notify，msg用來傳遞參數，notify用來通知外面這個協程任務已經執行完了。

在TestWorker方法中，我們模擬了10個任務，這10個任務不斷的往msg通道中發送數據，當msg有數據之后，我們創建的那10個協程就會爭取從msg通道接收消息，只要接收到消息就說明這是執行任務所必需的參數。執行完成之后向notify發送消息。在TestWorker中我們同樣接收了10次，在沒有消息的時候就會阻塞在<-notify這一行，直到有協程執行完成向notify通道發送消息，這里就從<-notify返回，進入下一次循環。上面其實就是一個非常簡單的協程池了，當然為了演示，代碼并不是很完整。

運行上面的代碼，得到的結果大概像下面這樣

2023/10/07 21:37:44 worker-id: 9
2023/10/07 21:37:44 worker-id: 4
2023/10/07 21:37:44 worker-id: 8
2023/10/07 21:37:44 m: 2
2023/10/07 21:37:44 m: 0
2023/10/07 21:37:44 worker-id: 5
2023/10/07 21:37:44 m: 3
2023/10/07 21:37:44 worker-id: 2
2023/10/07 21:37:44 worker-id: 1
2023/10/07 21:37:44 m: 5
2023/10/07 21:37:44 m: 4
2023/10/07 21:37:44 worker-id: 7
2023/10/07 21:37:44 m: 6
2023/10/07 21:37:44 worker-id: 0
2023/10/07 21:37:44 m: 7
2023/10/07 21:37:44 m: 1
2023/10/07 21:37:44 worker-id: 6
2023/10/07 21:37:44 m: 8
2023/10/07 21:37:44 m: 9
2023/10/07 21:37:44 worker-id: 3

從上面的結果來看，協程運行并不是順序執行的，這和多線程是一樣的道理。上面Golang的代碼執行的流程我畫了一張圖，如下：

注意箭頭的方向，所有協程都不斷的嘗試從channel中接收消息，拿到程序運行必要的參數，當msg中有數據時從case m := <-msg中蘇醒并執行具體的業務邏輯，我們知道，在Golang中channel是線程安全的，其內部有一把鎖，這把鎖就是mutex，下面是channel底層結構體

// src/runtime/chan.go:33
type hchan struct {
  ...
  lock mutex
}

channel除了能保證線程安全，還能保證順序性，也就是發送方最先發送的，在接收方一定也是最先收到的。這不就是一個加了鎖的隊列嗎？我們可以試著想一下在C++中是不是也可以實現類似的效果呢？不難想到，我們可以使用一個隊列在各個線程之間傳遞數據，像下面這樣：

主線程accept出來的套接字，只管往隊列里面丟就可以了，我們創建的一堆worker線程，不斷的嘗試從隊列里面pop數據。這樣，我們就解決了線程之間的交互問題。

下面，我們就參照上面Golang的代碼，先把這個框架給搭出來，然后再在這個基礎之上去完善代碼，最后實現一個準生產的線程池。

我們先參照上面Golang的代碼，實現相似邏輯，代碼如下：

#include 
#include 
#include 


void threadFunc(std::queue& q) {
    while (1) {
        if (!q.empty()) {
            int param = q.front();
            q.pop();
            std::cout << "param:" << param << std::endl;
        }
    }
}


void jobDispatch(std::queue& q) {
    for (int i = 0; i < 1000; i++) {
        q.push(i);
    }
}


int main() {
   std::queue q1;


   std::vector ths;
   for (int i = 0; i < 10; i++) {
       ths.emplace_back(threadFunc, std::ref(q1));
   }


   jobDispatch(q1);


   for (auto& th: ths) {
       th.join();
   }


   return 0;
}

上面的代碼盡可能的還原了Golang的邏輯，我們來分析一下這段代碼。在main函數中，創建了一個隊列q1，這個隊列用來向線程池傳遞參數，接著創建了10個線程保存在了vector中，將q1以引用的形式傳入線程池主函數(注意：這里傳引用必須使用std::ref包裝一下)，再接著調用jobDispatch模擬任務分配然后每個線程調用join等待結束。

接著我們來看線程池主函數threadFunc，這個函數接收一個隊列q作為參數，這里的q就是我們在創建線程池的時候傳進來的q1，然后是一個死循環，在這個循環里面我們不斷的判斷隊列是否為空，如果不為空就從隊列取出一個元素出來。最后，分配任務的函數jobDispatch向隊列q1里面push了1000個元素，來模擬1000個任務。

上面的代碼當然是有問題的，有興趣的可以把這段代碼拷貝下來把自己跑一下，你會發現雖然代碼能跑，但是結果完全看不懂。

首先，第一個問題就是queue不是線程安全的。所以，這個隊列得有一把鎖，比如：

std::mutex mtx;


void threadFunc(std::queue& q) {
    while (true) {
        if (!q.empty()) {
            std::lock_guard ltx(mtx);
            int param = q.front();
            q.pop();
            std::cout << "param:" << param << std::endl;
        }
    }
}

我們在threadFund函數中的出隊列之前加了一把鎖。這把鎖是全局的，每個線程都要先拿到這把鎖之后才能從隊列里拿到數據，你可以把這段代碼替換之后再運行一下，這次的結果應該是正確的了。

可能你覺得奇怪，我們使用lock_guard創建了一個ltx對象，但是并沒有地方去釋放這把鎖，如果你有這樣的疑問應該是對C++11還不是很熟悉，在C++11標準中，因為有了RAII的緣故，一個對象被銷毀時一定會執行析構函數，就算是運行過程中對象產生異常析構函數也會執行，在C++中這叫棧展開。有了這個特性之后，lock_guard就不難理解了，其構造函數其實就是調用了mutex的lock方法，然后把這個mutex保存在成員變量中，當對象銷毀時析構函數中調用unlock。所以，有了這個機制之后，我們就不用到處寫unlock了，這也是我覺得C++更好用了的原因之一。

在C++中同樣遵循大括號作用域，在上面的代碼中，lock_guard是在if語句中的，當if語句執行完之后，ltx就被銷毀了，所以當循環進入到下一次的時候實際上鎖已經被釋放了。

這樣我們就解決了隊列的線程安全問題，但眼尖的你一定看出來其實還有一個問題，threadFunc函數中的死循環一直在空轉，這顯然是有問題的。解決這個問題最容易想到的就是每次循環都sleep一下，但這顯然也是有問題的，對于一個有實時要求的系統是不能接受的。

所以，我們迫切需要一種機制，讓threadFunc沒事干的時候就停在那等通知，想想看什么技術可以實現？對，就是cond，在C++11中條件變量也被封裝在了std::命名空間中。下面我們就使用cond來改造一下，相關代碼如下：

std::mutex mtx;
std::condition_variable cond;   // v2


void threadFunc(std::queue& q) {
    while (true) {
        std::unique_lock ltx(mtx);       // v2
        cond.wait(ltx, [q]() { return !q.empty();}); // v2


        int param = q.front();
        q.pop();
        std::cout << "param:" << param << std::endl;
    }
}


void jobDispatch(std::queue& q) {
    for (int i = 0; i < 1000; i++) {
        q.push(i);
    }
    cond.notify_all();  // v2
}

修改后的代碼我在后面都加了注釋(v2)， 首先我們定義了一個全局的條件變量cond，然后在threadFunc中調用cond的wait方法等待通知。然后在jobDispatch中往隊列里面寫完數據之后調用notify_all方法通知所有等待的線程。這樣，當隊列中沒有數據的時候線程池中的線程就會進入睡眠，直到notify_all被調用。這里你可以想一下，上面notify_all還可以進一步優化嗎？

當然，上面還作了一個調整，就是將原來的lock_guard換了unique_lock，這個改動是必須的，我們知道cond調用wait的時候會嘗試釋放鎖，可lock_guard里面沒有釋放鎖的方法，而unique_lock是有unlock方法的。也就是說，unique_lock創建的ltx對象可以手動調用unlock方法釋放鎖。

好了，到這里其實我們已經寫出一個簡單的線程池了，這個線程池通過一個隊列傳遞參數，使用mutex解決線程安全問題，cond解決性能問題。看起來已經和上面Golang的邏輯非常接近了。如果你使用Golang寫過代碼，并且上面C++的代碼你也嘗試寫出來了，你就會驚嘆于Golang簡單了。好了，這里不吹Golang了，我們繼續回到C++上來。

當然，到這里還遠遠沒完呢，C++的看家本領是啥？對，是面向對象編程。上面的代碼很顯然沒有面向對象的味道。下面我們就使用面向對象的思想來實現一個線程池。這里直接給出代碼

#include 
#include 
#include 
#include 


class TPool {
public:
    TPool(): m_thread_size(1), m_terminate(false) {};
    ~TPool() { stop(); }
    // 線程池初始化
    bool init(size_t size);
    // 停止所有線程
    void stop();
    // 啟動線程池的各個線程
    void start();
    // 任務執行入口
    template 
    auto exec(F&& f, A&&... args)->std::future;
    // 等待所有線程執行完成
    bool waitDone();


private:
    // 每個任務都是一個struct結構體，方便未來擴展
    struct Task {
        Task() {}
        std::function m_func;
    };
    // Task未來在隊列中以智能指針的形式傳遞
    typedef std::shared_ptr TaskFunc;


private:
    // 嘗試從任務隊列獲取一個任務
    bool get(TaskFunc &t);
    // 線程主函數
    bool run();


private:
    // 任務隊列，將Task直接放到隊列中
    std::queue m_tasks;
    // 線程池
    std::vector m_threads;
    // 鎖
    std::mutex m_mutex;
    // 用于線程之間通知的條件變量
    std::condition_variable m_cond;
    // 線程池大小
    size_t m_thread_size;
    // 標記線程是否結束
    bool m_terminate;
    // 用來記錄狀態的原子變量
    std::atomic m_atomic{0};
};

我們定義了一個TPool類，這個類里面包含幾個部分，我們從下往上看。第一個部分是線程池管理相關的各種資源，每一個我都寫了注釋。第二部分是任務相關的操作，這部分不對外開放。第三部分是任務定義，使用struct聲明了一個Task，其中有一個空的構造函數，還聲明了一個m_func，這是最終task執行的入口。最后一部分是線程池對外開放的各種接口。用戶使用線程的大致流程如下：

這里我將線程的初始化和線程的啟動分成了兩步，是希望在使用的時候精確知道是哪一步出了問題，如果你覺得這樣太繁瑣，可以適當減少步驟，比如將init和start方法進行合并。

下面我們就來詳細講一下線程各個方法的實現。首先是init和start方法，init方法用來初始化線程，代碼如下：

bool init(size_t size) {
    unique_lock lock(m_mutex);
    if (!m_threads.empty()) {
        return false;
    }
    m_thread_size = size;
    return true;
}

傳入一個size，表示線程池的大小，上來就加鎖，這是為了防止在多線程的情景下執行init，這個方法實際上只做了一件事，就是設置線程池的大小。

初始化完了之后，調用start方法啟動線程池，start方法的代碼如下：

bool start() {
    unique_lock lock(m_mutex);
    if (!m_threads.empty()) {
        return false;
    }


    for (size_t i = 0; i < m_thread_size; i++) {
        m_threads.push_back(new thread(&TPool::run, this));
    }
    return true;
}

void run() {
    while(!m_terminate) {
        TaskFunc task;
        bool ok = get(task);
        if (ok) {
            ++m_atomic;


            task->m_func();


            --m_atomic;


            unique_lock lock(m_mutex);
            if (m_atomic == 0 && m_tasks.empty()) { // 是否所有任務都執行完成
                m_cond.notify_all();
            }
        }
    }
}

不出所料，run方法里其實就是一個死循環，這個循環上來就判斷是否結束了，如果已經結束就退出當前循環，run方法返回，當前線程結束。

如果沒有結束，就調用get方法從任務隊列里取一個任務出來執行，這里使用一個原子變量來判斷最后是不是所有任務都執行完成，這個原子變量不是必須的，你可以根據你自己的場景做相應的修改。取到任務之后，就會調用任務的m_func方法，還記得這個方法嗎？它定義在Task結構體中。最后會判斷是否所有任務都結束了，如果已經結束了會通知其它線程。

這里我們來看一下get方法是怎么獲取任務的，get方法的代碼如下：

bool get(TaskFunc &t) {
    unique_lock lock(m_mutex);
    if (m_tasks.empty()) {
        m_cond.wait(lock, [this]{return m_terminate || !m_tasks.empty();});
    }


    if (m_terminate)
        return false;


    t = std::move(m_tasks.front());
    m_tasks.pop();
    return true;
}

上來首先加了一把鎖，如果任務隊列沒有任務可以執行，使用條件變量m_cond調用wait方法等待。

然后，如果此時線程已經被結束掉了，直接返回false，如果沒有結束，就從隊列中取出一個任務，賦值給傳進來的t。注意，這里使用的是參數傳值的方式。這樣就實現了任務的傳遞，當沒有任務的時候m_cond.wait會讓當前進程進入睡眠，等待通知。

接下來，我們看一下任務是如何被投遞到任務隊列中的，用來投遞任務的方法是exec，代碼如下：

template 
auto exec(F&& f, A&&... args)->future {
    using retType = decltype(f(args...));
    auto task = make_shared>(bind(std::forward(f), std::forward(args)...));
    TaskFunc fPtr = make_shared();
    fPtr->m_func = [task](){
        (*task)();
    };


    unique_lock lock(m_mutex);
    m_tasks.push(fPtr);
    m_cond.notify_one();


    return task->get_future();
}

using retType = decltype(f(args...));

autotask=make_shared>(bind(std::forward(f),std::forward(args)...));

TaskFunc fPtr = make_shared();
fPtr->m_func = [task](){
    (*task)();
};

void stop() {
    {
        unique_lock lock(m_mutex);
        m_terminate = true;
        m_cond.notify_all();
    }


    for (auto & m_thread : m_threads) {
        if (m_thread->joinable()) {
            m_thread->join();
        }
        delete m_thread;
        m_thread = nullptr;
    }


    unique_lock lock(m_mutex);
    m_threads.clear();
}

#include "thread-pool.hpp"
#include 


using namespace std;


void threadFunc(int a) {
   cout << "a=" << a << endl;
}


class A {
public:
    A() = default;
    int run(int a, int b) {
        return a + b;
    }
};


int main() {
    TPool p1;
    p1.init(10);
    p1.start();
    p1.exec(threadFunc, 100);
    p1.exec(threadFunc, 200);


    A a1;
    auto fu1 = p1.exec(std::bind(&A::run, &a1, std::_1, std::_2), 10, 20);
    int ret = fu1.get();
    std::cout << "res:" << ret << std::endl;


    p1.waitDone();
    return 0;
}