手擼一個線程池

點擊上方藍字關(guān)注下我唄

之前分享過一次手寫線程池 - C語言版，然后有朋友問是否有C++線程池實現(xiàn)的文章：

其實關(guān)于C++線程池的文章我好久以前寫過，但估計很多新朋友都沒有看到過，這里也重新發(fā)一下！

本人在開發(fā)過程中經(jīng)常會遇到需要使用線程池的需求，但查了一圈發(fā)現(xiàn)在C++中完備的線程池第三方庫還是比較少的，于是打算自己搞一個，鏈接地址文章最后附上，目前還只是初版，可能還有很多問題，望各位指正。

個人認為線程池需要支持以下幾個基本功能：

• 核心線程數(shù)(core_threads)：線程池中擁有的最少線程個數(shù)，初始化時就會創(chuàng)建好的線程，常駐于線程池。

• 最大線程個數(shù)(max_threads)：線程池中擁有的最大線程個數(shù)，max_threads>=core_threads，當(dāng)任務(wù)的個數(shù)太多線程池執(zhí)行不過來時，內(nèi)部就會創(chuàng)建更多的線程用于執(zhí)行更多的任務(wù)，內(nèi)部線程數(shù)不會超過max_threads，多創(chuàng)建出來的線程在一段時間內(nèi)沒有執(zhí)行任務(wù)則會自動被回收掉，最終線程個數(shù)保持在核心線程數(shù)。

• 超時時間(time_out)：如上所述，多創(chuàng)建出來的線程在time_out時間內(nèi)沒有執(zhí)行任務(wù)就會被回收。

• 可獲取當(dāng)前線程池中線程的總個數(shù)。

• 可獲取當(dāng)前線程池中空閑線程的個數(shù)。

• 開啟線程池功能的開關(guān)。

• 關(guān)閉線程池功能的開關(guān)，可以選擇是否立即關(guān)閉，立即關(guān)閉線程池時，當(dāng)前線程池里緩存的任務(wù)不會被執(zhí)行。

如何實現(xiàn)線程池？下面是自己實現(xiàn)的線程池邏輯。

1. 鏈表或者數(shù)組：用于存儲線程池中的線程。

2. 隊列：用于存儲需要放入線程池中執(zhí)行的任務(wù)。

3. 條件變量：當(dāng)有任務(wù)需要執(zhí)行時，用于通知正在等待的線程從任務(wù)隊列中取出任務(wù)執(zhí)行。

代碼如下：

class ThreadPool { public:  using PoolSeconds = std::chrono::seconds;
  /** 線程池的配置   * core_threads: 核心線程個數(shù)，線程池中最少擁有的線程個數(shù)，初始化就會創(chuàng)建好的線程，常駐于線程池   *   * max_threads: >=core_threads，當(dāng)任務(wù)的個數(shù)太多線程池執(zhí)行不過來時，   * 內(nèi)部就會創(chuàng)建更多的線程用于執(zhí)行更多的任務(wù)，內(nèi)部線程數(shù)不會超過max_threads   *   * max_task_size: 內(nèi)部允許存儲的最大任務(wù)個數(shù)，暫時沒有使用   *   * time_out: Cache線程的超時時間，Cache線程指的是max_threads-core_threads的線程,   * 當(dāng)time_out時間內(nèi)沒有執(zhí)行任務(wù)，此線程就會被自動回收   */  struct ThreadPoolConfig {      int core_threads;      int max_threads;      int max_task_size;      PoolSeconds time_out;  };
  /**   * 線程的狀態(tài)：有等待、運行、停止   */  enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 };
  /**   * 線程的種類標(biāo)識：標(biāo)志該線程是核心線程還是Cache線程，Cache是內(nèi)部為了執(zhí)行更多任務(wù)臨時創(chuàng)建出來的   */  enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 };
  using ThreadPtr = std::shared_ptr<std::thread>;  using ThreadId = std::atomic<int>;  using ThreadStateAtomic = std::atomic;  using ThreadFlagAtomic = std::atomic;
  /**   * 線程池中線程存在的基本單位，每個線程都有個自定義的ID，有線程種類標(biāo)識和狀態(tài)   */  struct ThreadWrapper {      ThreadPtr ptr;      ThreadId id;      ThreadFlagAtomic flag;      ThreadStateAtomic state;
      ThreadWrapper() {          ptr = nullptr;          id = 0;          state.store(ThreadState::kInit);      }  };  using ThreadWrapperPtr = std::shared_ptr;  using ThreadPoolLock = std::unique_lock<std::mutex>;
 private:  ThreadPoolConfig config_;
  std::list worker_threads_;
  std::queue<std::function<void()>> tasks_;  std::mutex task_mutex_;  std::condition_variable task_cv_;
  std::atomic<int> total_function_num_;  std::atomic<int> waiting_thread_num_;  std::atomic<int> thread_id_; // 用于為新創(chuàng)建的線程分配ID
  std::atomic<bool> is_shutdown_now_;  std::atomic<bool> is_shutdown_;  std::atomic<bool> is_available_;};

在構(gòu)造函數(shù)中將各個成員變量都附初值，同時判斷線程池的config是否合法。

ThreadPool(ThreadPoolConfig config) : config_(config) {    this->total_function_num_.store(0);    this->waiting_thread_num_.store(0);
    this->thread_id_.store(0);    this->is_shutdown_.store(false);    this->is_shutdown_now_.store(false);
    if (IsValidConfig(config_)) {        is_available_.store(true);    } else {        is_available_.store(false);    }}
bool IsValidConfig(ThreadPoolConfig config) {    if (config.core_threads < 1 || config.max_threads < config.core_threads || config.time_out.count() < 1) {        return false;    }    return true;}

創(chuàng)建核心線程數(shù)個線程，常駐于線程池，等待任務(wù)的執(zhí)行，線程ID由GetNextThreadId()統(tǒng)一分配。

// 開啟線程池功能bool Start() {    if (!IsAvailable()) {        return false;    }    int core_thread_num = config_.core_threads;    cout << "Init thread num " << core_thread_num << endl;    while (core_thread_num-- > 0) {        AddThread(GetNextThreadId());    }    cout << "Init thread end" << endl;    return true;}

這里有兩個標(biāo)志位，is_shutdown_now置為true表示立即關(guān)閉線程，is_shutdown置為true則表示先執(zhí)行完隊列里的任務(wù)再關(guān)閉線程池。

// 關(guān)掉線程池，內(nèi)部還沒有執(zhí)行的任務(wù)會繼續(xù)執(zhí)行void ShutDown() {    ShutDown(false);    cout << "shutdown" << endl;}
// 執(zhí)行關(guān)掉線程池，內(nèi)部還沒有執(zhí)行的任務(wù)直接取消，不會再執(zhí)行void ShutDownNow() {    ShutDown(true);    cout << "shutdown now" << endl;}
// privatevoid ShutDown(bool is_now) {    if (is_available_.load()) {        if (is_now) {            this->is_shutdown_now_.store(true);        } else {            this->is_shutdown_.store(true);        }        this->task_cv_.notify_all();        is_available_.store(false);    }}

見AddThread()函數(shù)，默認會創(chuàng)建Core線程，也可以選擇創(chuàng)建Cache線程，線程內(nèi)部會有一個死循環(huán)，不停的等待任務(wù)，有任務(wù)到來時就會執(zhí)行，同時內(nèi)部會判斷是否是Cache線程，如果是Cache線程，timeout時間內(nèi)沒有任務(wù)執(zhí)行就會自動退出循環(huán)，線程結(jié)束。

這里還會檢查is_shutdown和is_shutdown_now標(biāo)志，根據(jù)兩個標(biāo)志位是否為true來判斷是否結(jié)束線程。

void AddThread(int id) { AddThread(id, ThreadFlag::kCore); }
void AddThread(int id, ThreadFlag thread_flag) {    cout << "AddThread " << id << " flag " << static_cast<int>(thread_flag) << endl;    ThreadWrapperPtr thread_ptr = std::make_shared();    thread_ptr->id.store(id);    thread_ptr->flag.store(thread_flag);    auto func = [this, thread_ptr]() {        for (;;) {            std::function<void()> task;            {                ThreadPoolLock lock(this->task_mutex_);                if (thread_ptr->state.load() == ThreadState::kStop) {                    break;                }                cout << "thread id " << thread_ptr->id.load() << " running start" << endl;                thread_ptr->state.store(ThreadState::kWaiting);                ++this->waiting_thread_num_;                bool is_timeout = false;                if (thread_ptr->flag.load() == ThreadFlag::kCore) {                    this->task_cv_.wait(lock, [this, thread_ptr] {                        return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||                                thread_ptr->state.load() == ThreadState::kStop);                    });                } else {                    this->task_cv_.wait_for(lock, this->config_.time_out, [this, thread_ptr] {                        return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||                                thread_ptr->state.load() == ThreadState::kStop);                    });                    is_timeout = !(this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||                                    thread_ptr->state.load() == ThreadState::kStop);                }                --this->waiting_thread_num_;                cout << "thread id " << thread_ptr->id.load() << " running wait end" << endl;
                if (is_timeout) {                    thread_ptr->state.store(ThreadState::kStop);                }
                if (thread_ptr->state.load() == ThreadState::kStop) {                    cout << "thread id " << thread_ptr->id.load() << " state stop" << endl;                    break;                }                if (this->is_shutdown_ && this->tasks_.empty()) {                    cout << "thread id " << thread_ptr->id.load() << " shutdown" << endl;                    break;                }                if (this->is_shutdown_now_) {                    cout << "thread id " << thread_ptr->id.load() << " shutdown now" << endl;                    break;                }                thread_ptr->state.store(ThreadState::kRunning);                task = std::move(this->tasks_.front());                this->tasks_.pop();            }            task();        }        cout << "thread id " << thread_ptr->id.load() << " running end" << endl;    };    thread_ptr->ptr = std::make_shared<std::thread>(std::move(func));    if (thread_ptr->ptr->joinable()) {        thread_ptr->ptr->detach();    }    this->worker_threads_.emplace_back(std::move(thread_ptr));}

見如下代碼，將任務(wù)使用std::bind封裝成std::function放入任務(wù)隊列中，任務(wù)較多時內(nèi)部還會判斷是否有空閑線程，如果沒有空閑線程，會自動創(chuàng)建出最多(max_threads-core_threads)個Cache線程用于執(zhí)行任務(wù)。

// 放在線程池中執(zhí)行函數(shù)template auto Run(F &&f, Args &&... args) -> std::shared_ptr>> {    if (this->is_shutdown_.load() || this->is_shutdown_now_.load() || !IsAvailable()) {        return nullptr;    }    if (GetWaitingThreadSize() == 0 && GetTotalThreadSize() < config_.max_threads) {        AddThread(GetNextThreadId(), ThreadFlag::kCache);    }
    using return_type = std::result_of_t;    auto task = std::make_shared>(        std::bind(std::forward(f), std::forward(args)...));    total_function_num_++;
    std::future res = task->get_future();    {        ThreadPoolLock lock(this->task_mutex_);        this->tasks_.emplace([task]() { (*task)(); });    }    this->task_cv_.notify_one();    return std::make_shared>>(std::move(res));}

int GetTotalThreadSize() { return this->worker_threads_.size(); }

waiting_thread_num值表示空閑線程的個數(shù)，該變量在線程循環(huán)內(nèi)部會更新。

int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }

int main() {    cout << "hello" << endl;    ThreadPool pool(ThreadPool::ThreadPoolConfig{4, 5, 6, std::chrono::seconds(4)});    pool.Start();    std::this_thread::sleep_for(std::chrono::seconds(4));    cout << "thread size " << pool.GetTotalThreadSize() << endl;    std::atomic<int> index;    index.store(0);    std::thread t([&]() {        for (int i = 0; i < 10; ++i) {            pool.Run([&]() {                cout << "function " << index.load() << endl;                std::this_thread::sleep_for(std::chrono::seconds(4));                index++;            });            // std::this_thread::sleep_for(std::chrono::seconds(2));        }    });    t.detach();    cout << "=================" << endl;
    std::this_thread::sleep_for(std::chrono::seconds(4));    pool.Reset(ThreadPool::ThreadPoolConfig{4, 4, 6, std::chrono::seconds(4)});    std::this_thread::sleep_for(std::chrono::seconds(4));    cout << "thread size " << pool.GetTotalThreadSize() << endl;    cout << "waiting size " << pool.GetWaitingThreadSize() << endl;    cout << "---------------" << endl;    pool.ShutDownNow();    getchar();    cout << "world" << endl;    return 0;}

關(guān)于線程池個人認為還應(yīng)該有定時器功能和循環(huán)執(zhí)行某個任務(wù)的功能，這兩個功能我是單獨封裝成一個類實現(xiàn)，可以點擊閱讀原文。

后臺回復(fù) “C++線程池” ，獲取完整代碼。

往期推薦

new[]和delete[]一定要配對使用嗎？

分享一個編程設(shè)計小技巧（沒有兩三年工作經(jīng)驗估計看不懂）

多線程學(xué)習(xí)指南

if-else和switch-case哪個效率更高？看這四張圖。

從未見過把內(nèi)存玩的如此明白的文章（推薦大家都來看看）

寫出高效代碼的12條建議

Linux C++ 服務(wù)器端這條線怎么走？

關(guān)于多線程，我給出13點建議

shared_ptr是線程安全的嗎？

累夠嗆！整理了一份C++學(xué)習(xí)路線圖！

推薦學(xué)習(xí)這個C++開源項目

普通的int main(){}沒有寫return 0;會怎么樣？

點個在看你最好看