線程池的C++實現

這個代碼不是本人原創，而是網上的代碼 https://github.com/progschj/ThreadPool

1. 大致思路

線程池目的是減少創建銷毀線程的開銷。大致的思想是生產者消費者模型，主線程為生產者，負責往任務隊列中加新任務，如果沒有新任務則發出結束信號。消費者線程不停檢查任務隊列和結束信號，如果有任務則取一個處理。沒有則等待，如果檢測到結束信號則退出。

剩下的問題是，消費者處理完任務的返回值如何存放。在這個實現中使用了std::future用於等待和接收任務的返回值。我們先來看一下std::future是如何用作等待函數的返回值的

#include <iostream>
#include <future>
 
int main() {
    // 封裝一下函數
    std::packaged_task<int()> task([]{ return 7; }); 
    //獲取返回值place_holder
    std::future<int> f1 = task.get_future();  
    //執行函數
    task(); 
    f1.wait();
    //打印 7
    std::cout <<  f1.get()  << '\n';  
    return 0;
}

直到task();之前，函數都沒有執行，只是將函數體和函數返回值作了分離：task封裝了函數體，f1封裝了返回值。按照這個思路，生產者將task的函數體放入到任務隊列中，供消費者調用，而將所有任務的返回值作為std::future類型放入到一個數組中，供外部程序訪問即可。

2. 代碼注釋

現在可以看代碼並思考一些細節了

//ThreadPool.h
#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

class ThreadPool {
public:
    ThreadPool(size_t);
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>;
    ~ThreadPool();
private:
    // 消費者線程
    std::vector< std::thread > workers;     
    //任務隊列
    std::queue< std::function<void()> > tasks;  
    
    //互斥量
    std::mutex queue_mutex;              
    std::condition_variable condition;
    //停止信號，如果為true，則表示沒有新的任務
    bool stop;                 
};
 
//構造函數，分配threads個消費者線程
inline ThreadPool::ThreadPool(size_t threads)     
    :   stop(false)
{
    for(size_t i = 0;i<threads;++i)
        workers.emplace_back(
            [this]
            {
                //無限循環
                for(;;)                
                {
                    //取出的任務放這里
                    std::function<void()> task;           

                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        //檢查任務隊列和停止信號，如果有任務或者有停止信號則往下執行，否則釋放鎖讓其他線程可以訪問鎖，並將自己阻塞
                        this->condition.wait(lock,
                            [this]{ return this->stop || !this->tasks.empty(); });  
                        //如果收到了停止信號，並且任務隊列為空，則退出當前線程
                        if(this->stop && this->tasks.empty())   
                            return;
                        //否則取出一個任務
                        task = std::move(this->tasks.front());   
                        this->tasks.pop();
                    }
                    //執行取出的任務
                    task();   
                }
            }
        );
}


//生產者添加一個新的任務，其函數類型為F，參數類型為Args，這里的具名右值引用是為了統一const T &和T &兩種形參，算是形參的萬能模版
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type>  
{
    //std::result_of<F(Args...)>::type是函數的返回值類型
    using return_type = typename std::result_of<F(Args...)>::type;

    //下面std::forward的作用是為了保持具名右值引用，否則原來的T &&傳入到task時會變成T &
    //使用bind讓所有形參消失，從而統一了任務的函數類型
    //使用shared_ptr是因為后面要復制這個指針到任務隊列中
    auto task = std::make_shared< std::packaged_task<return_type()> >(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)  
                    
        );
    //分離返回值和函數體
    std::future<return_type> res = task->get_future();    
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        //如果已經停止，則不能加入新的任務
        if(stop)                            
            throw std::runtime_error("enqueue on stopped ThreadPool");
        //這里*task就是std::packaged_task<return_type()>，(*task)() = f(args...)，而tasks是std::function，封裝了這段執行的代碼，真正的執行是在消費者線程中
        tasks.emplace([task](){ (*task)(); });     
    }
    //通知一個消費者線程，有新任務了
    condition.notify_one();
    return res;
}


inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        //沒有更多任務了
        stop = true;  
    }
    //通知所有消費者線程，拿到任務的執行剩下的任務，沒有任務的退出
    condition.notify_all();  
    //結束所有消費者線程
    for(std::thread &worker: workers)
        worker.join();     
}

#endif

來看調用代碼

#include <iostream>
#include <vector>
#include <chrono>

#include "ThreadPool.h"

int main()
{
    //4個消費者，沒有任務時，處於阻塞狀態
    ThreadPool pool(4);   
    //任務返回值
    std::vector< std::future<int> > results; 

    //8個任務
    for(int i = 0; i < 8; ++i) { 
        results.emplace_back(
            pool.enqueue([i] {
                std::cout << "hello " << i << std::endl;
                std::this_thread::sleep_for(std::chrono::seconds(1));
                std::cout << "world " << i << std::endl;
                return i*i;
            })
        );
    }
    //獲取返回值
    for(auto && result: results)
        std::cout << result.get() << ' ';  
    std::cout << std::endl;
    
    return 0;
}

應該來說還是很清楚簡單的。