手擼一個線程池
[導讀]點擊上方藍字關注下我唄之前分享過一次手寫線程池-C語言版，然后有朋友問是否有C線程池實現的文章：其實關于C線程池的文章我好久以前寫過，但估計很多新朋友都沒有看到過，這里也重新發(fā)一下！本人在開發(fā)過程中經常會遇到需要使用線程池的需求，但查了一圈發(fā)現在C中完備的線程池第三方庫還是比較少...
點擊上方藍字關注下我唄
之前分享過一次手寫線程池 - C語言版，然后有朋友問是否有C 線程池實現的文章：
其實關于C 線程池的文章我好久以前寫過，但估計很多新朋友都沒有看到過，這里也重新發(fā)一下！本人在開發(fā)過程中經常會遇到需要使用線程池的需求，但查了一圈發(fā)現在C 中完備的線程池第三方庫還是比較少的，于是打算自己搞一個，鏈接地址文章最后附上，目前還只是初版，可能還有很多問題，望各位指正。
線程池都需要什么功能？
個人認為線程池需要支持以下幾個基本功能：
- 核心線程數(core_threads)：線程池中擁有的最少線程個數，初始化時就會創(chuàng)建好的線程，常駐于線程池。
- 最大線程個數(max_threads)：線程池中擁有的最大線程個數，max_threads>=core_threads，當任務的個數太多線程池執(zhí)行不過來時，內部就會創(chuàng)建更多的線程用于執(zhí)行更多的任務，內部線程數不會超過max_threads，多創(chuàng)建出來的線程在一段時間內沒有執(zhí)行任務則會自動被回收掉，最終線程個數保持在核心線程數。
- 超時時間(time_out)：如上所述，多創(chuàng)建出來的線程在time_out時間內沒有執(zhí)行任務就會被回收。
- 可獲取當前線程池中線程的總個數。
- 可獲取當前線程池中空閑線程的個數。
- 開啟線程池功能的開關。
- 關閉線程池功能的開關，可以選擇是否立即關閉，立即關閉線程池時，當前線程池里緩存的任務不會被執(zhí)行。
如何實現線程池？下面是自己實現的線程池邏輯。
線程池中主要的數據結構
1. 鏈表或者數組：用于存儲線程池中的線程。2. 隊列：用于存儲需要放入線程池中執(zhí)行的任務。3. 條件變量：當有任務需要執(zhí)行時，用于通知正在等待的線程從任務隊列中取出任務執(zhí)行。代碼如下：
class ThreadPool { public: using PoolSeconds = std::chrono::seconds;
/** 線程池的配置 * core_threads: 核心線程個數，線程池中最少擁有的線程個數，初始化就會創(chuàng)建好的線程，常駐于線程池 * * max_threads: >=core_threads，當任務的個數太多線程池執(zhí)行不過來時， * 內部就會創(chuàng)建更多的線程用于執(zhí)行更多的任務，內部線程數不會超過max_threads * * max_task_size: 內部允許存儲的最大任務個數，暫時沒有使用 * * time_out: Cache線程的超時時間，Cache線程指的是max_threads-core_threads的線程, * 當time_out時間內沒有執(zhí)行任務，此線程就會被自動回收 */ 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 };
/** * 線程的種類標識：標志該線程是核心線程還是Cache線程，Cache是內部為了執(zhí)行更多任務臨時創(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，有線程種類標識和狀態(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_;};線程池的初始化在構造函數中將各個成員變量都附初值，同時判斷線程池的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)建核心線程數個線程，常駐于線程池，等待任務的執(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;}如何關閉線程？這里有兩個標志位，is_shutdown_now置為true表示立即關閉線程，is_shutdown置為true則表示先執(zhí)行完隊列里的任務再關閉線程池。
// 關掉線程池，內部還沒有執(zhí)行的任務會繼續(xù)執(zhí)行void ShutDown() { ShutDown(false); cout << "shutdown" << endl;}
// 執(zhí)行關掉線程池，內部還沒有執(zhí)行的任務直接取消，不會再執(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()函數，默認會創(chuàng)建Core線程，也可以選擇創(chuàng)建Cache線程，線程內部會有一個死循環(huán)，不停的等待任務，有任務到來時就會執(zhí)行，同時內部會判斷是否是Cache線程，如果是Cache線程，timeout時間內沒有任務執(zhí)行就會自動退出循環(huán)，線程結束。
這里還會檢查is_shutdown和is_shutdown_now標志，根據兩個標志位是否為true來判斷是否結束線程。
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_ 
下載文檔
