C++實(shí)現(xiàn)簡單的線程池

線程池編程簡介：

????在我們的服務(wù)端的程序中運(yùn)用了大量關(guān)于池的概念，線程池、連接池、內(nèi)存池、對象池等等。使用池的概念后可以高效利用服務(wù)器端的資源，比如沒有大量的線程在系統(tǒng)中進(jìn)行上下文的切換，一個數(shù)據(jù)庫連接池，也只需要維護(hù)一定里的連接，而不是占用很多數(shù)據(jù)庫連接資源。同時它們也避免了一些耗時的操作，比如創(chuàng)建一個線程，申請一個數(shù)據(jù)庫連接，而且可能就只使用那么一次，然后就立刻釋放剛申請的資源，效率很低。

????在我的上一篇blog中已經(jīng)實(shí)現(xiàn)一個線程基類了，在這里我們只需要實(shí)現(xiàn)一個線程池類ThreadPool和該線程池調(diào)度的工作線程類WorkThread即可，而且WorkThread是繼承自Thread類的。

?

實(shí)現(xiàn)思路：

????一個簡單的線程池的實(shí)現(xiàn)思路一般如下：

在ThreadPool中創(chuàng)建多個線程(WorkThreadk對象)，每個線程均處于阻塞狀態(tài)，等待任務(wù)的到來

ThreadPool提供一個提交任務(wù)的接口，如post_job(ProcCallback func, void* data); post_job后會立即返回，不會阻塞

ThreadPool維護(hù)一個空閑線程隊(duì)列，當(dāng)客戶程序調(diào)用post_job()后，如果空閑隊(duì)列中有空閑線程，則取出一個線程句柄，并設(shè)置任務(wù)再給出新任務(wù)通知事件即可，處理等待的線程捕捉到事件信號后便開始執(zhí)行任務(wù)，執(zhí)行完后將該線程句柄重新push到空閑線程隊(duì)列中

該線程池采用回調(diào)函數(shù)方式

首先我們實(shí)現(xiàn)一個WorkThread類：

?1?typedef?void?(APR_THREAD_FUNC?*ProcCallBack)(void*);????????//回調(diào)函數(shù)指針
?2?//由線程池調(diào)度的工作線程
?3?class?WorkThread?:?public?Thread ? ? ?//Thread類的實(shí)現(xiàn)可參考我上一篇的blog: 《C++封裝一個簡單的線程類》
?4?{
?5?????friend?class?ThreadPool;
?6?public:
?7?????WorkThread(ThreadPool*?pthr_pool)
?8?????{
?9?????????thr_pool_????=?pthr_pool;
10?????????cb_func_????=?NULL;
11?????????param_????????=?NULL;
12?????}
13?????virtual?~WorkThread(){}
14?????void????set_job(ProcCallBack?func,?void*?param)
15?????{
16?????????cb_func_????=?func;
17?????????param_????????=?param;
18?????????notify();????????????//通知有新的任務(wù)
19?????}
20?????//實(shí)現(xiàn)Thread的run方法，并調(diào)用用戶指定的函數(shù)
21?????virtual?void?run()
22?????{
23?????????if?(cb_func_)
24?????????????cb_func_(param_);
25?????
26?????????//reset?callback?function?pointer
27?????????cb_func_????=?NULL;
28?????????param_????????=?NULL;
29?????
30?????????//執(zhí)行完任務(wù)，將該線程句柄移到線程池空閑隊(duì)列
31?????????thr_pool_->move_to_idle_que(this);
32?????}
33?????
34?private:
35?????ThreadPool*????thr_pool_;????????//線程池指針
36?????ProcCallBack????cb_func_;????????//回調(diào)函數(shù)地址
37?????void*? ? ? ? ? ? ?param_; ? ? ? ? ??//回調(diào)函數(shù)參數(shù)
38?};

該WorkThread中，有一個回調(diào)函數(shù)指針和參數(shù)，當(dāng)有新任務(wù)時，會在run()中被調(diào)用，執(zhí)行完后會將該線程移動到空閑線程隊(duì)列，等待下一次任務(wù)的提交。

ThreadPool類定義如下：

?1?class?ThreadPool
?2?{
?3?????friend?class?WorkThread;
?4?public:
?5?????ThreadPool();
?6?????virtual?~ThreadPool();
?7?????int????start_thread_pool(size_t?thread_num?=?5);????????//啟動thread_num個線程
?8?????int????stop_thread_pool();??????????????????????????????????????//線束線程池
?9?????void????destroy();????????????????????????????????????????????????//銷毀線程池所申請的資源
10?????void????post_job(ProcCallBack?func,?void*?data);????????//提交任務(wù)接口，傳入回調(diào)函數(shù)地址和參數(shù)
11?
12?protected:
13?????WorkThread*????get_idle_thread();??????????????????????????????//從獲得空閑隊(duì)列中取得一個線程句柄
14?????void????????append_idle_thread(WorkThread*?pthread);????//加入到thread_vec_和idl_que_中
15?????void????????move_to_idle_que(WorkThread*?idlethread);????//將線程句柄加入到idle_que_中
16?
17?private:
18?????size_t????????????????thr_num_;??????????????????????//線程數(shù)目
19?????vector<WorkThread*>????????thr_vec_;????????//線程句柄集合
20?????BlockQueue<WorkThread*>????idle_que_; ? ??//空閑線程隊(duì)列
21?
22?private:
23?????//?not?implement
24?????ThreadPool(const?ThreadPool&?);
25?????ThreadPool&????operator=(const?ThreadPool&?);
26?}; 線程池實(shí)現(xiàn)的關(guān)鍵是如何創(chuàng)建多個線程，并且當(dāng)任務(wù)來臨時可以從線程池中取一個線程（也就是去得到其中一個線程的指針），然后提交任務(wù)并執(zhí)行。還有一點(diǎn)就是當(dāng)任務(wù)執(zhí)行完后，應(yīng)該將該線程句柄重新加入到空閑線程隊(duì)列，所以我們將ThreadPool的指針傳入給了WorkThread，thr_pool_最后可以調(diào)用move_to_idle_que(this)來將該線程句柄移到空閑隊(duì)列中。 ThreadPool中一些關(guān)鍵代碼的實(shí)現(xiàn)： ?1?int?ThreadPool::start_thread_pool(size_t?thread_num)
?2?{
?3?????assert(thread_num?!=?0);
?4?????thr_num_????=?thread_num;
?5?????int????ret????????=?0;
?6?????for?(size_t?i?=?0;?i?<?thr_num_;?++i)
?7?????{
?8?????????WorkThread*????pthr?=?new?WorkThread(this);
?9?????????pthr->set_thread_id(i);
10?????????if?((ret?=?pthr->start())?!=?0)
11?????????{
12?????????????printf("start_thread_pool:?failed?when?create?a?work?thread:?%d\n",?i);
13?????????????delete?pthr;
14?????????????return?i;
15?????????}
16?????????append_idle_thread(pthr);????????
17?????}
18?????return?thr_num_;
19?}
20?int?ThreadPool::stop_thread_pool()
21?{
22?????for?(size_t?i?=?0;?i?<?thr_vec_.size();?++i)
23?????{
24?????????WorkThread*?pthr?=?thr_vec_[i];
25?????????pthr->join();
26?????????delete?pthr;
27?????}
28?????thr_vec_.clear();
29?????idle_que_.clear();
30?????return?0;
31?}
32?void?ThreadPool::destroy()
33?{
34?????stop_thread_pool();
35?}
36?void?ThreadPool::append_idle_thread(WorkThread*?pthread)
37?{
38?????thr_vec_.push_back(pthread);
39?????idle_que_.push(pthread);
40?}
41?void?ThreadPool::move_to_idle_que(WorkThread*?idlethread)
42?{
43?????idle_que_.push(idlethread);
44?}
45?WorkThread*?ThreadPool::get_idle_thread()
46?{
47?????WorkThread*????pthr?=?NULL;
48?????if?(!idle_que_.empty())
49?????????pthr?=?idle_que_.take();
50?????return?pthr;
51?}
52?void?ThreadPool::post_job(ProcCallBack?func,?void*?data)
53?{
54?????assert(func?!=?NULL);
55?????WorkThread*?pthr?=?get_idle_thread();
56?????while?(pthr?==?NULL)
57?????{
58?????????apr_sleep(500000);
59?????????pthr?=?get_idle_thread();
60?????}
61?????pthr->set_job(func,?data);
62?} ThreadPool中的BlockQueue<WorkThread*> 也就是一個線程安全的隊(duì)列，即對std::deque做了一個包裝，在插入和取出元素時加了一個讀寫鎖。

使用示例： //任務(wù)執(zhí)行函數(shù)，必須是ProcCallback類型 void count(void* param) { // do some your work, like:? int* pi = static_cast<int*>(param); int val = *pi + 1; printf("val=%d\n", val); pelete pi; } //程序中使用如下： ThreadPool* ptp = new ThreadPool(); ptp->start_thread_pool(3); //啟動3 個線程 ptp->post_job(count, new int(1)); //提交任務(wù) ptp->post_job(count, new int(2)); ptp->post_job(count, new int(3)); //程序線束時 ptp->stop_thread_pool(); 其實(shí)count()函數(shù)就是我們的業(yè)務(wù)實(shí)現(xiàn)代碼，有任務(wù)時，可以提交給線程池去執(zhí)行。 結(jié)尾： 其實(shí)實(shí)現(xiàn)一個線程池或其它什么池并不難，當(dāng)然線程安全和效率還是要從多寫代碼的經(jīng)驗(yàn)中獲取。像這個線程池也就是基于預(yù)創(chuàng)多個建線程，保保存好它們的線程句柄，當(dāng)有新任務(wù)時取一個線程執(zhí)行即可，執(zhí)行完后一定要?dú)w還到空閑線程隊(duì)列中，當(dāng)然我們可以在線程池中增加一個任務(wù)隊(duì)列，因?yàn)楫?dāng)post_job()時，若當(dāng)時沒有空閑線程，有兩種方案，一是等待有空閑線程，二是加入到任務(wù)隊(duì)列，當(dāng)WorkThread線程執(zhí)行完一個任務(wù)后，從任務(wù)隊(duì)列中取一個任務(wù)繼續(xù)執(zhí)行即可，不會阻塞在post_job()中。 另外，我們可以封裝一些線程安全的隊(duì)列和map什么的，這樣在程序中就不用擔(dān)心創(chuàng)建一個多線程共享的隊(duì)列時，還必須創(chuàng)建一個鎖，挺麻煩的，比如上面的BlockQueue<Type>直接拿來用就行了。

轉(zhuǎn)載于:https://www.cnblogs.com/huty/p/8517387.html

總結(jié)

以上是生活随笔為你收集整理的【C/C++开发】C++实现简单的线程池的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。