一、引言

本文目標(biāo)：從零實現(xiàn)一個輕量、可用、可擴(kuò)展的 C++ 線程池。

關(guān)鍵技術(shù)點：

1. std::thread
2. std::mutex / std::unique_lock
3. std::condition_variable
4. std::function / future / packaged_task / bind

二、什么是線程池

線程池是一種用于管理和復(fù)用線程的并發(fā)編程模型。它的核心思想是預(yù)先創(chuàng)建一組工作線程，并將它們放入一個“池”中進(jìn)行管理。當(dāng)有新任務(wù)需要處理時，不再創(chuàng)建和銷毀線程，而是直接將任務(wù)提交給線程池，由池中空閑的線程來執(zhí)行。

三、為什么需要線程池

頻繁地創(chuàng)建和銷毀線程會帶來顯著的系統(tǒng)開銷，包括內(nèi)存分配、切換到內(nèi)核態(tài)等。線程池通過復(fù)用線程，有效解決了這個問題，并帶來了以下優(yōu)勢：

• 降低資源開銷：避免了頻繁創(chuàng)建和銷毀線程的開銷，提升了系統(tǒng)性能。
• 提高響應(yīng)速度：任務(wù)到達(dá)時無需等待線程創(chuàng)建，可以立即被分配給空閑線程執(zhí)行。
• 提高線程的可管理性：線程是稀缺資源，無限制地創(chuàng)建線程會消耗大量系統(tǒng)內(nèi)存，甚至導(dǎo)致內(nèi)存溢出。線程池可以對線程數(shù)量進(jìn)行統(tǒng)一分配、調(diào)優(yōu)和監(jiān)控。

四、線程池的核心組成

1. 工作線程集合 (Worker Threads)：池中預(yù)先創(chuàng)建好的一組線程，它們會持續(xù)運行，不斷從任務(wù)隊列中獲取并執(zhí)行任務(wù)。
2. 任務(wù)隊列 (Task Queue)：一個線程安全的隊列，用于存放所有待執(zhí)行的任務(wù)。它作為任務(wù)提交者和工作線程之間的緩沖區(qū)。
3. 同步機(jī)制 (Synchronization)：
   • 互斥鎖 (Mutex)：用于保護(hù)任務(wù)隊列，確保在多線程環(huán)境下對隊列的訪問是安全的，防止競態(tài)條件。
   • 條件變量 (Condition Variable)：用于工作線程的等待和喚醒。當(dāng)任務(wù)隊列為空時，工作線程會進(jìn)入等待狀態(tài)；當(dāng)有新任務(wù)加入時，會通知（喚醒）一個或所有等待的線程。
4. 任務(wù)接口 (Task Interface)：一個用于提交任務(wù)的方法，允許外部將各種類型的任務(wù)（函數(shù)、Lambda表達(dá)式等）提交到線程池中。

五、C++線程池的實現(xiàn)

因為在代碼中使用到了一些異步編程技術(shù)，所以先做個簡單的介紹。

std::condition_variable——條件變量，它是一種線程間的同步機(jī)制，當(dāng)沒有任務(wù)時，它會阻塞工作線程。生產(chǎn)者線程將任務(wù)加入隊列后，會通過同一個條件變量喚醒在該條件變量下等待的線程。這么做的好處是，避免了工作線程循環(huán)檢測隊列中有沒有任務(wù)帶來的CPU開銷。代碼中用到的接口主要有3個：

void wait (unique_lock<mutex>& lck, Predicate pred);

第一個參數(shù)是互斥鎖，該函數(shù)內(nèi)部會將鎖釋放，避免線程休眠時持有鎖，導(dǎo)致其他活躍線程拿不到鎖。

第二個參數(shù)是一個可調(diào)用對象，這個可調(diào)用對象必須能夠返回true或false。而且，這個可調(diào)用對象會循環(huán)的執(zhí)行，直到它的返回結(jié)果是true。

pred可調(diào)用對象返回true后，線程被喚醒，重新獲取鎖，向下執(zhí)行。

void notify_one() noexcept;

喚醒一個在該條件變量下等待的線程。

void notify_all() noexcept;

喚醒所有在該條件變量下等待的線程。

std::future，用來獲取異步執(zhí)行的結(jié)果。如果沒有std::future，在C++中想要獲取其他線程的返回值的話，我們需要將這個返回值寫入全局變量，這樣其他的線程才可以看到。全局變量是共享資源，多線程場景下是需要加鎖保護(hù)的，std::future封裝了這些底層的細(xì)節(jié)，提供了一種同步獲取結(jié)果的方式。

get()方法

會阻塞調(diào)用線程，直到拿到結(jié)果才繼續(xù)往下執(zhí)行，所以說它是一種同步獲取結(jié)果的方式。

std::function，是一種函數(shù)包裝器，統(tǒng)一函數(shù)的類型。為什么這么說？函數(shù)指針、lambda表達(dá)式雖然都是可調(diào)用對象，但是它們的類型是完全不同的，所以就不能夠放到同一個容器中統(tǒng)一管理。但是，經(jīng)過function的包裝后，它們就有了統(tǒng)一的類型。下面在實現(xiàn)線程池中就可以看到如何使用。

std::packaged_task，是一種任務(wù)包裝器，用來獲取異步執(zhí)行任務(wù)的結(jié)果的。它內(nèi)部有一個關(guān)聯(lián)的future對象，函數(shù)的返回值會被寫入到這個future對象中，同時它還提供了一個獲取這個future對象的接口get_future()，其他線程拿到這個關(guān)聯(lián)的future后，就可以調(diào)用get()方法獲取異步執(zhí)行的結(jié)果。順便說一下，packaged_task內(nèi)部重載了（），可以直接通過packaged_task對象執(zhí)行它包裝的任務(wù)。

std::bind，綁定函數(shù)參數(shù)，并返回一個可調(diào)用對象。比如，函數(shù)Add(int a, int b)原本是需要傳入兩個參數(shù)的，是以這種形式 Add(10, 20) 調(diào)用的。但是，經(jīng)過std::bind綁定后，例如 auto func = std::bind(Add, 10, 20); 在調(diào)用的時候，就不用傳參了，直接這樣 func() 調(diào)用，效果是一樣的。為什么要綁定參數(shù)，在下面的線程池實現(xiàn)中，就很明白了。

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <future>
#include <mutex>
#include <functional>
#include <chrono>
#include <type_traits>
#include <condition_variable>
class ThreadPool {
public:
    ThreadPool(size_t thread_num = 4):
        _thread_num(thread_num),
        _start(false),
        _stop(false)
    {}
    ~ThreadPool(){
        if (_start && !_stop) stop();
    }
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool& operator=(ThreadPool&&) = delete;
    void start() {
        std::unique_lock<std::mutex> lock(_mutex);
        if (_start) return;
        _workers.reserve(_thread_num);
        for (size_t i = 0; i < _thread_num; i++) {
            _workers.emplace_back(std::thread([this](){
                work_loop();
            }));
        }
        _start = true;
    }
    void stop() {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_start || _stop) return;
            // 在join回收線程之前，必須先將_stop置為true，
            // 否則工作線程可能會一直阻塞在條件變量上，導(dǎo)致無法正常退出，甚至?xí)?dǎo)致程序崩潰
            _stop = true;
        }
        _cond.notify_all();
        for (auto& worker : _workers) {
            if (worker.joinable()) worker.join();
        }
    }
    template<class F, class... Args>
    auto submit(F&& f, Args&&... args)->std::future<std::invoke_result_t<F, Args...>> {
        using return_type = std::invoke_result_t<F, Args...>;
        // 綁定函數(shù)參數(shù)，并交給任務(wù)包裝器
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        // 獲取關(guān)聯(lián)的future
        std::future<return_type> res = task->get_future();
        // 加鎖+入隊列
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_stop || !_start) throw std::runtime_error("線程池未啟動！");
            _tasks.emplace([task](){(*task)();});
        }
        _cond.notify_one();
        return res;
    }
private:
    void work_loop() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock, [this](){
                    return _stop || !_tasks.empty();
                });
                if (_stop && _tasks.empty()) return;
                task = std::move(_tasks.front());
                _tasks.pop();
            }
            task();
        }
    }
private:
    std::vector<std::thread> _workers;
    std::queue<std::function<void()>> _tasks;
    std::mutex _mutex;
    std::condition_variable _cond;
    size_t _thread_num;
    bool _start;
    bool _stop;
};
int add(int a, int b) {
    return a + b;
}
void print() {
    std::cout << "-------------------print-------------------" << std::endl;
    std::cout << "Hello World!" << std::endl;
}
int main() {
    ThreadPool pool(4);
    pool.start();
    std::cout << "==================ThreadPoolTest==================" << std::endl;
    pool.submit([](){
        std::cout << "-------------------lambda-------------------" << std::endl;
        std::cout << "this is a lambda!" << std::endl;
    });
    std::this_thread::sleep_for(std::chrono::seconds(3));  
    auto ret1 = pool.submit(add, 10, 20);
    std::cout << "-------------------add-------------------" << std::endl;
    std::cout << "10 + 20 = " << ret1.get() << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(3));  
    pool.submit(print);
    pool.stop();
    return 0;
}

運行結(jié)果：

在上述代碼實現(xiàn)中，用到了C++17的語法。

這里我解釋一下隊列中的任務(wù)參數(shù)為空，并且返回值為void，但是帶返回值的add函數(shù)為什么可以插入隊列中。

通過std::bind綁定函數(shù)的所有參數(shù)，所有就做到了“無參”。

通過std::packaged_task獲取函數(shù)的返回值，所以不擔(dān)心返回值拿不到。

通過lambda封裝一層，不管原本是否有參數(shù)，是否有返回值，加入隊列中的任務(wù)都是滿足無參和無返回值的。

這三個操作組合在一起，不管函數(shù)有無返回值，都能適配到任務(wù)隊列中。

另外，在綁定參數(shù)時，不僅對函數(shù)參數(shù)進(jìn)行了完美轉(zhuǎn)發(fā)，還對函數(shù)本身進(jìn)行了完美轉(zhuǎn)發(fā)。對參數(shù)進(jìn)行完美轉(zhuǎn)發(fā)是因為要保持它本身的左右值屬性。對函數(shù)進(jìn)行完美轉(zhuǎn)發(fā)是因為，我們在使用時，可能會直接在submit函數(shù)傳入lambda，這時候lambda它是一個右值，采用完美轉(zhuǎn)發(fā)可以保持它的右值屬性，觸發(fā)移動語義，也就避免了std::bind內(nèi)部對它進(jìn)行拷貝。

以上就是從零實現(xiàn)一個輕量級C++線程池的詳細(xì)內(nèi)容，更多關(guān)于輕量級C++線程池實現(xiàn)的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論