哈嘍，我是老吳，我又來分享學(xué)習(xí)心得了。另外，為了更好地體現(xiàn)公眾號(hào)的核心價(jià)值觀，從本文開始，我會(huì)在文末新增分享一些非技術(shù)相關(guān)的內(nèi)容，歡迎大家參與討論。

一、簡介

https://github.com/Pithikos/C-Thread-Pool

這是一個(gè)簡單小巧的C語言線程池實(shí)現(xiàn)，在 Github 上有 1.1K 的 star，很適合用來學(xué)習(xí) Linux 的多線程編程。

另外，里面還涉及到了信號(hào)、隊(duì)列、同步等知識(shí)點(diǎn)，代碼讀起來還是挺過癮的。

特點(diǎn)：

• 符合 ANCI C and POSIX;
• 支持暫停/恢復(fù)/等待功能;
• 簡潔的 API;
• 經(jīng)過嚴(yán)格的測試，附帶了豐富的測試用例;

二、使用

快速上手

example.c：

#include?"thpool.h"

void?task(void?*arg){
?printf("Thread?#%u?working?on?%d\n",?(int)pthread_self(),?(int)?arg);
}

int?main(){
?
?puts("Making?threadpool?with?4?threads");
?threadpool?thpool?=?thpool_init(4);

?puts("Adding?10?tasks?to?threadpool");
?int?i;
?for?(i=0;?i<8;?i ){
??thpool_add_work(thpool,?task,?(void*)(uintptr_t)i);
?};

?thpool_wait(thpool);
?puts("Killing?threadpool");
?thpool_destroy(thpool);
?
?return?0;
}
運(yùn)行效果：

$?gcc?example.c?thpool.c?-D?THPOOL_DEBUG?-pthread?-o?example

$?./example
Making?threadpool?with?4?threads
THPOOL_DEBUG:?Created?thread?0?in?pool?
THPOOL_DEBUG:?Created?thread?1?in?pool?
THPOOL_DEBUG:?Created?thread?2?in?pool?
THPOOL_DEBUG:?Created?thread?3?in?pool?
Adding?10?tasks?to?threadpool
Thread?#1509455616?working?on?0
Thread?#1509455616?working?on?4
Thread?#1509455616?working?on?5
Thread?#1492670208?working?on?2
Thread?#1492670208?working?on?7
Thread?#1509455616?working?on?6
Thread?#1501062912?working?on?1
Thread?#1517848320?working?on?3
Killing?threadpool
代碼分析：

• threadpool thpool = thpool_init(4) 創(chuàng)建了一個(gè)含有 4 個(gè)線程的線程池;
• 然后調(diào)用 thpool_add_work(thpool, ...) 往線程池里放入了 8 個(gè)任務(wù);
• 從結(jié)果來看：
• 線程5616 搶到了任務(wù) 0 / 4 / 5 / 6;
• 線程0208 搶到了任務(wù) 2 / 7;
• 線程2919 搶到了任務(wù) 1;
• 線程8320 搶到了任務(wù) 3;

API 簡介

示例	作用
thpool_init(4)	創(chuàng)建一個(gè)含有 4 個(gè)線程的線程池。
thpool_add_work(thpool, (void*)function_p, (void*)arg_p)	添加任務(wù)， function_p 是任務(wù)要執(zhí)行的函數(shù)，arg_p 是 function_p 的參數(shù)。
thpool_wait(thpool)	等待所有任務(wù)完成。
thpool_destroy(thpool)	銷毀線程池，如果還有任務(wù)在執(zhí)行，則會(huì)先等待其完成。
thpool_pause(thpool)	讓所有的線程都停止工作，進(jìn)入睡眠狀態(tài)。
thpool_resume(thpool)	讓所有的線程都恢復(fù)工作。
thpool_num_threads_working(thpool)	返回當(dāng)前正在工作的線程數(shù)。

三、內(nèi)部實(shí)現(xiàn)

整體把握

核心代碼就是 2 個(gè)文件：thpool.c 和 thpool.h。

分解 thpool.c

7 個(gè)公共函數(shù)：

struct?thpool_*?thpool_init(int?num_threads)?
int?thpool_add_work(thpool_*?thpool_p,?void?(*function_p)(void*),?void*?arg_p)?
void?thpool_wait(thpool_*?thpool_p)?
void?thpool_destroy(thpool_*?thpool_p)?
void?thpool_pause(thpool_*?thpool_p)?
void?thpool_resume(thpool_*?thpool_p)?
int?thpool_num_threads_working(thpool_*?thpool_p)?
正好就是前面說過的 7 個(gè) API，稍后重點(diǎn)分析。

5 個(gè)自定義的數(shù)據(jù)結(jié)構(gòu)：

//?描述一個(gè)信號(hào)量
typedef?struct?bsem?{...}?bsem;

//?描述一個(gè)任務(wù)
typedef?struct?job?{...}?job;

//?描述一個(gè)任務(wù)隊(duì)列
typedef?struct?jobqueue?{...}?jobqueue;

//?描述一個(gè)線程
typedef?struct?thread?{...}?thread;

//?描述一個(gè)線程池
typedef?struct?thpool_?{...}?thpool_;
14 個(gè)私有函數(shù)：

//?構(gòu)造?struct?thread，并調(diào)用?pthread_create()?創(chuàng)建線程
static?int?thread_init?(thpool_*?thpool_p,?struct?thread**?thread_p,?int?id)?

//?當(dāng)線程被暫停時(shí)會(huì)在這里休眠
static?void?thread_hold(int?sig_id)?

//?線程在此函數(shù)中執(zhí)行任務(wù)
static?void*?thread_do(struct?thread*?thread_p)?

//?銷毀?struct?thread
static?void?thread_destroy?(thread*?thread_p)?

//?任務(wù)隊(duì)列相關(guān)的操作集合
static?int?jobqueue_init(jobqueue*?jobqueue_p)?
static?void?jobqueue_clear(jobqueue*?jobqueue_p)?
static?void?jobqueue_push(jobqueue*?jobqueue_p,?struct?job*?newjob)?
static?struct?job*?jobqueue_pull(jobqueue*?jobqueue_p)?
static?void?jobqueue_destroy(jobqueue*?jobqueue_p)?

//?信號(hào)量相關(guān)的操作集合
static?void?bsem_init(bsem?*bsem_p,?int?value)?
static?void?bsem_reset(bsem?*bsem_p)?
static?void?bsem_post(bsem?*bsem_p)?
static?void?bsem_post_all(bsem?*bsem_p)?
static?void?bsem_wait(bsem*?bsem_p)


核心 API 的實(shí)現(xiàn)

1. thpool_init()

該函數(shù)用于創(chuàng)建一個(gè)線程池，先明確線程池的定義：

typedef?struct?thpool_{
?thread**???threads;??????????????????/*?pointer?to?threads????????*/
?volatile?int?num_threads_alive;??????/*?threads?currently?alive???*/
?volatile?int?num_threads_working;????/*?threads?currently?working?*/
?pthread_mutex_t??thcount_lock;???????/*?used?for?thread?count?etc?*/
?pthread_cond_t??threads_all_idle;????/*?signal?to?thpool_wait?????*/
?jobqueue??jobqueue;??????????????????/*?job?queue?????????????????*/
}?thpool_;
thpool_init() 的實(shí)現(xiàn)思路：

1. 分配 struct thpool_：
• malloc(sizeof(struct thpool_))
2. 初始化 struct thpool_;
• malloc(num_threads * sizeof(struct thread *))
• thread_init(thpool_p,