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

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

在我看來，除了統(tǒng)一C++編程接口之外，C++11標(biāo)準(zhǔn)更多的是在語言層面做了很多優(yōu)化，規(guī)避了原來C語言中的很多陷阱，比如C++11中的lock_guard、future、promise等技術(shù)，將原來C語言中語法上容易犯錯(cuò)的地方進(jìn)行了簡化，簡單來說就是將原來依賴人的地方交給了編譯器（很多時(shí)候機(jī)器比人更可靠）。比如在C++11標(biāo)準(zhǔn)之前，我們使用mutex是像下面這樣的：

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

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

C++編譯器會(huì)自動(dòng)為我們插入釋放鎖的代碼，這樣我們就不用時(shí)刻擔(dān)心鎖有沒有正確釋放了。我個(gè)人的感覺是，從C++11標(biāo)準(zhǔn)開始，C++變得不那么可怕了，甚至很多時(shí)候覺得C++變得好用了。

這篇文章我們試著使用C++11標(biāo)準(zhǔn)的新特性來實(shí)現(xiàn)一個(gè)通用的線程池。首先，我們來看一下C++11標(biāo)準(zhǔn)中線程是如何創(chuàng)建的，先看代碼：

#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創(chuàng)建一個(gè)t1線程對象，傳入線程主函數(shù)，將x以引用的方式傳入線程主函數(shù)，接著調(diào)用join方法等待主函數(shù)threadFunc執(zhí)行完成。上面x最終的結(jié)果將是20。

整個(gè)流程和使用pthread_create其實(shí)區(qū)別不大，只是C++11將線程的創(chuàng)建封裝成了一個(gè)類，使得我們可以用面向?qū)ο缶幊痰姆绞絹韯?chuàng)建多線程。

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

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

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

為了說明這個(gè)問題，我們先來看一段Golang的代碼，暫時(shí)看不懂沒關(guān)系，后面我會(huì)解釋

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


  buildWorkers(poolSize, msg, notify)


  // 模擬任務(wù)
  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方法創(chuàng)建了10個(gè)協(xié)程，每個(gè)協(xié)程里面都有一個(gè)for循環(huán)，一開始每個(gè)for循環(huán)都阻塞在select語句中的case m := ←msg，如果之前沒有接觸過Go語言（這里的select你可以簡單和Linux中的select技術(shù)類比）。另外還有兩個(gè)通道，一個(gè)是msg和notify，msg用來傳遞參數(shù)，notify用來通知外面這個(gè)協(xié)程任務(wù)已經(jīng)執(zhí)行完了。

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

運(yùn)行上面的代碼，得到的結(jié)果大概像下面這樣

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

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

注意箭頭的方向，所有協(xié)程都不斷的嘗試從channel中接收消息，拿到程序運(yùn)行必要的參數(shù)，當(dāng)msg中有數(shù)據(jù)時(shí)從case m := <-msg中蘇醒并執(zhí)行具體的業(yè)務(wù)邏輯，我們知道，在Golang中channel是線程安全的，其內(nèi)部有一把鎖，這把鎖就是mutex，下面是channel底層結(jié)構(gòu)體

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

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

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

下面，我們就參照上面Golang的代碼，先把這個(gè)框架給搭出來，然后再在這個(gè)基礎(chǔ)之上去完善代碼，最后實(shí)現(xiàn)一個(gè)準(zhǔn)生產(chǎn)的線程池。

我們先參照上面Golang的代碼，實(shí)現(xiàn)相似邏輯，代碼如下：

#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函數(shù)中，創(chuàng)建了一個(gè)隊(duì)列q1，這個(gè)隊(duì)列用來向線程池傳遞參數(shù)，接著創(chuàng)建了10個(gè)線程保存在了vector中，將q1以引用的形式傳入線程池主函數(shù)(注意：這里傳引用必須使用std::ref包裝一下)，再接著調(diào)用jobDispatch模擬任務(wù)分配然后每個(gè)線程調(diào)用join等待結(jié)束。

接著我們來看線程池主函數(shù)threadFunc，這個(gè)函數(shù)接收一個(gè)隊(duì)列q作為參數(shù)，這里的q就是我們在創(chuàng)建線程池的時(shí)候傳進(jìn)來的q1，然后是一個(gè)死循環(huán)，在這個(gè)循環(huán)里面我們不斷的判斷隊(duì)列是否為空，如果不為空就從隊(duì)列取出一個(gè)元素出來。最后，分配任務(wù)的函數(shù)jobDispatch向隊(duì)列q1里面push了1000個(gè)元素，來模擬1000個(gè)任務(wù)。

上面的代碼當(dāng)然是有問題的，有興趣的可以把這段代碼拷貝下來把自己跑一下，你會(huì)發(fā)現(xiàn)雖然代碼能跑，但是結(jié)果完全看不懂。

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

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函數(shù)中的出隊(duì)列之前加了一把鎖。這把鎖是全局的，每個(gè)線程都要先拿到這把鎖之后才能從隊(duì)列里拿到數(shù)據(jù)，你可以把這段代碼替換之后再運(yùn)行一下，這次的結(jié)果應(yīng)該是正確的了。

可能你覺得奇怪，我們使用lock_guard創(chuàng)建了一個(gè)ltx對象，但是并沒有地方去釋放這把鎖，如果你有這樣的疑問應(yīng)該是對C++11還不是很熟悉，在C++11標(biāo)準(zhǔn)中，因?yàn)橛辛薘AII的緣故，一個(gè)對象被銷毀時(shí)一定會(huì)執(zhí)行析構(gòu)函數(shù)，就算是運(yùn)行過程中對象產(chǎn)生異常析構(gòu)函數(shù)也會(huì)執(zhí)行，在C++中這叫棧展開。有了這個(gè)特性之后，lock_guard就不難理解了，其構(gòu)造函數(shù)其實(shí)就是調(diào)用了mutex的lock方法，然后把這個(gè)mutex保存在成員變量中，當(dāng)對象銷毀時(shí)析構(gòu)函數(shù)中調(diào)用unlock。所以，有了這個(gè)機(jī)制之后，我們就不用到處寫unlock了，這也是我覺得C++更好用了的原因之一。

在C++中同樣遵循大括號作用域，在上面的代碼中，lock_guard是在if語句中的，當(dāng)if語句執(zhí)行完之后，ltx就被銷毀了，所以當(dāng)循環(huán)進(jìn)入到下一次的時(shí)候?qū)嶋H上鎖已經(jīng)被釋放了。

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

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

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)， 首先我們定義了一個(gè)全局的條件變量cond，然后在threadFunc中調(diào)用cond的wait方法等待通知。然后在jobDispatch中往隊(duì)列里面寫完數(shù)據(jù)之后調(diào)用notify_all方法通知所有等待的線程。這樣，當(dāng)隊(duì)列中沒有數(shù)據(jù)的時(shí)候線程池中的線程就會(huì)進(jìn)入睡眠，直到notify_all被調(diào)用。這里你可以想一下，上面notify_all還可以進(jìn)一步優(yōu)化嗎？

當(dāng)然，上面還作了一個(gè)調(diào)整，就是將原來的lock_guard換了unique_lock，這個(gè)改動(dòng)是必須的，我們知道cond調(diào)用wait的時(shí)候會(huì)嘗試釋放鎖，可lock_guard里面沒有釋放鎖的方法，而unique_lock是有unlock方法的。也就是說，unique_lock創(chuàng)建的ltx對象可以手動(dòng)調(diào)用unlock方法釋放鎖。

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

當(dāng)然，到這里還遠(yuǎn)遠(yuǎn)沒完呢，C++的看家本領(lǐng)是啥？對，是面向?qū)ο缶幊獭Ｉ厦娴拇a很顯然沒有面向?qū)ο蟮奈兜馈Ｏ旅嫖覀兙褪褂妹嫦驅(qū)ο蟮乃枷雭韺?shí)現(xiàn)一個(gè)線程池。這里直接給出代碼

#include 
#include 
#include 
#include 


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


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


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


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

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

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

下面我們就來詳細(xì)講一下線程各個(gè)方法的實(shí)現(xiàn)。首先是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;
}

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

初始化完了之后，調(diào)用start方法啟動(dòng)線程池，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()) { // 是否所有任務(wù)都執(zhí)行完成
                m_cond.notify_all();
            }
        }
    }
}

不出所料，run方法里其實(shí)就是一個(gè)死循環(huán)，這個(gè)循環(huán)上來就判斷是否結(jié)束了，如果已經(jīng)結(jié)束就退出當(dāng)前循環(huán)，run方法返回，當(dāng)前線程結(jié)束。

如果沒有結(jié)束，就調(diào)用get方法從任務(wù)隊(duì)列里取一個(gè)任務(wù)出來執(zhí)行，這里使用一個(gè)原子變量來判斷最后是不是所有任務(wù)都執(zhí)行完成，這個(gè)原子變量不是必須的，你可以根據(jù)你自己的場景做相應(yīng)的修改。取到任務(wù)之后，就會(huì)調(diào)用任務(wù)的m_func方法，還記得這個(gè)方法嗎？它定義在Task結(jié)構(gòu)體中。最后會(huì)判斷是否所有任務(wù)都結(jié)束了，如果已經(jīng)結(jié)束了會(huì)通知其它線程。

這里我們來看一下get方法是怎么獲取任務(wù)的，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;
}

上來首先加了一把鎖，如果任務(wù)隊(duì)列沒有任務(wù)可以執(zhí)行，使用條件變量m_cond調(diào)用wait方法等待。

然后，如果此時(shí)線程已經(jīng)被結(jié)束掉了，直接返回false，如果沒有結(jié)束，就從隊(duì)列中取出一個(gè)任務(wù)，賦值給傳進(jìn)來的t。注意，這里使用的是參數(shù)傳值的方式。這樣就實(shí)現(xiàn)了任務(wù)的傳遞，當(dāng)沒有任務(wù)的時(shí)候m_cond.wait會(huì)讓當(dāng)前進(jìn)程進(jìn)入睡眠，等待通知。

接下來，我們看一下任務(wù)是如何被投遞到任務(wù)隊(duì)列中的，用來投遞任務(wù)的方法是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;
}