C 語言版執行緒池
一、初始執行緒池
1.1 何為執行緒池？
我們先來打個比方，執行緒池就好像一個工具箱，我們每次需要擰螺絲的時候都要從工具箱裡面取出一個螺絲刀來。有時候需要取出一個來擰，有時候螺絲多的時候需要多個人取出多個來擰，擰完自己的螺絲那麼就會把螺絲刀再放回去，然後別人下次用的時候再取出來用。
說白了執行緒池就是相當於「提前申請了一些資源，也就是執行緒」，需要的時候就從執行緒池中取出執行緒來處理一些事情，處理完畢之後再把執行緒放回去。
1.2 為什麼要使用執行緒池？
我們來思考一個問題，為什麼需要執行緒池呢？假如沒有執行緒池的話我們每次呼叫執行緒是什麼樣子的？
顯然首先是先建立一個執行緒，然後把任務交給這個執行緒，最後把這個執行緒銷燬掉。這樣實現起來非常簡便，但是就會有一個問題：如果並行的執行緒數量很多，並且每個執行緒都是執行一個時間很短的任務就結束了，這樣頻繁建立執行緒就會大大降低系統的效率，因為頻繁建立執行緒和銷燬執行緒是需要消耗時間的。
那麼如果我們改用執行緒池的話，在程式執行的時候就會首先建立一批執行緒，然後交給執行緒池來管理。有需要的時候我們從執行緒池中取出執行緒用於處理任務，用完後我們再將其回收到執行緒池中，這樣是不是就避免了每次都需要建立和銷燬執行緒這種耗時的操作。
有人會說你使用執行緒池一開始就消耗了一些記憶體，之後一直不釋放這些記憶體，這樣豈不是有點浪費。其實這是類似於空間換時間的概念，我們確實多佔用了一點記憶體但是這些記憶體和我們珍惜出來的這些時間相比，是非常划算的。
池的概念是一種非常常見的空間換時間的概念，除了有執行緒池之外還有程序池、記憶體池等等。其實他們的思想都是一樣的就是我先申請一批資源出來，然後就隨用隨拿，不用再放回來。
1.3 如何設計執行緒池
執行緒池的組成主要分為 3 個部分，這三部分配合工作就可以得到一個完整的執行緒池：
- 任務佇列：儲存需要處理的任務，由工作的執行緒來處理這些任務。
- 通過執行緒池提供的 API 函數，將一個待處理的任務新增到任務佇列，或者從任務佇列中刪除
- 已處理的任務會被從任務佇列中刪除
- 執行緒池的使用者，也就是呼叫執行緒池函數往任務佇列中新增任務的執行緒就是生產者執行緒
- 工作的執行緒（任務佇列任務的消費者）：若干個，一般情況下根據 CPU 的核數來確定。
- 執行緒池中維護了一定數量的工作執行緒，他們的作用是不停的讀任務佇列，從裡邊取出任務並處理
- 工作的執行緒相當於是任務佇列的消費者角色
- 如果任務佇列為空，工作的執行緒將會被阻塞 (使用條件變數 / 號誌阻塞)
- 如果阻塞之後有了新的任務，由生產者將阻塞解除，工作執行緒開始工作
- 管理者執行緒（不處理任務佇列中的任務）：1 個
- 它的任務是週期性的對任務佇列中的任務數量以及處於忙狀態的工作執行緒個數進行檢測
- 當任務過多的時候，可以適當的建立一些新的工作執行緒
- 當任務過少的時候，可以適當的銷燬一些工作的執行緒
二、C 語言版執行緒池
由於本篇是對執行緒池的簡單介紹，所以簡化了一下執行緒池的模型，將 1.3 中的「3. 管理者執行緒」的角色給去除了。
2.1 結構體定義
2.1.1 任務結構體
/* 任務結構體 */
typedef struct
{
void (*function)(void *);
void *arg;
} threadpool_task_t;
2.1.2 執行緒池結構體
/* 執行緒池結構體 */
typedef struct
{
pthread_mutex_t lock; // 執行緒池鎖，鎖整個的執行緒池
pthread_cond_t notify; // 條件變數，用於告知執行緒池中的執行緒來任務了
int thread_count; // 執行緒池中的工作執行緒總數
pthread_t *threads; // 執行緒池中的工作執行緒
int started; // 執行緒池中正在工作的執行緒個數
threadpool_task_t *queue; // 任務佇列
int queue_size; // 任務佇列能容納的最大任務數
int head; // 隊頭 -> 取任務
int tail; // 隊尾 -> 放任務
int count; // 任務佇列中剩餘的任務個數
int shutdown; // 執行緒池狀態, 0 表示執行緒池可用，其餘值表示關閉
} threadpool_t;
-
thread_count 和 started 的區別：
- 初始化執行緒池的時候會建立一批執行緒（假設建立 n 個），此時 thread_count = started = n
- 當執行緒池執行過程中可能需要關閉一些執行緒（假設關閉 m 個），則會銷燬這些執行緒，並 started -= n，但 thread_count 保持不變
- 即 thread_count 表示執行緒池中的申請的執行緒個數，而 started 表示當前能用的執行緒個數
-
shutdown 的作用：如果需要銷燬執行緒池，那麼必須要現將所有的執行緒退出才可銷燬，而 shutdown 就是用於告知正在工作中的執行緒，執行緒池是否關閉用的。關閉方式又分為兩種：一種是立即關閉，即不管任務佇列中是否還有任務；另一種是優雅的關閉，即先處理完任務佇列中的任務後再關閉。這兩種方式可通過設定 shutdown 的不同取值即可實現：
typedef enum { immediate_shutdown = 1, // 立即關閉執行緒池 graceful_shutdown = 2 // 等執行緒池中的任務全部處理完成後，再關閉執行緒池 } threadpool_shutdown_t;
2.2 函數定義
2.2.1 ThreadPool_Init
函數原型：int ThreadPool_Init(int thread_count, int queue_size, threadpool_t **ppstThreadPool);
頭 文 件：#include "ThrdPool.h"
函數功能：初始化執行緒池
引數描述：
- thread_count：入參，代表此次建立的執行緒池中的執行緒個數
- queue_size：入參，代表任務佇列大小
- ppstThreadPool：出參，如果建立成功，則代表建立好的執行緒池，否則為 NULL
返 回 值：成功返回 E_SUCCEED，失敗返回 E_ERROR
2.2.2 ThreadPool_Dispatch
函數原型：int ThreadPool_Dispatch(threadpool_t *pstThreadPool, void (*function)(void *), void *arg);
頭 文 件：#include "ThrdPool.h"
函數功能：向執行緒池的任務佇列中分發任務
引數描述：
- pstThreadPool：入參，代表建立好的執行緒池
- function：入參，表示任務
- arg：入參，代表 function 的引數
返 回 值：成功返回 E_SUCCEED，失敗返回 E_ERROR
2.2.2 Threadpool_Destroy
函數原型：void Threadpool_Destroy(threadpool_t *pool, threadpool_shutdown_t shutdown_mode);
頭 文 件：#include "ThrdPool.h"
函數功能：銷燬執行緒池
引數描述：
- pool：入參，表示需要銷燬的執行緒池
- shutdown_mode：入參，表示關閉模式，有兩種取值
2.3 原始碼
2.3.1 ThrdPool.h
#ifndef __THRDPOOL_H__
#define __THRDPOOL_H__
#include <pthread.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#define DEBUG(format, args...) \
printf("[%s:%d] "format"\n", \
__FILE__, \
__LINE__, \
##args)
#define MAX_THREADS 16 // 執行緒池最大工作執行緒個數
#define MAX_QUEUE 256 // 執行緒池工作佇列上限
#define E_SUCCEED 0
#define E_ERROR 112
#define SAFE_FREE(ptr) \
if (ptr) \
{ \
free(ptr); \
ptr = NULL; \
}
/* 任務結構體 */
typedef struct
{
void (*function)(void *);
void *arg;
} threadpool_task_t;
/* 執行緒池結構體 */
typedef struct
{
pthread_mutex_t lock; // 執行緒池鎖，鎖整個的執行緒池
pthread_cond_t notify; // 條件變數，用於告知執行緒池中的執行緒來任務了
int thread_count; // 執行緒池中的工作執行緒總數
pthread_t *threads; // 執行緒池中的工作執行緒
int started; // 執行緒池中正在工作的執行緒個數
threadpool_task_t *queue; // 任務佇列
int queue_size; // 任務佇列能容納的最大任務數
int head; // 隊頭 -> 取任務
int tail; // 隊尾 -> 放任務
int count; // 任務佇列中剩餘的任務個數
int shutdown; // 執行緒池狀態, 0 表示執行緒池可用，其餘值表示關閉
} threadpool_t;
typedef enum
{
immediate_shutdown = 1, // 立即關閉執行緒池
graceful_shutdown = 2 // 等執行緒池中的任務全部處理完成後，再關閉執行緒池
} threadpool_shutdown_t;
int ThreadPool_Init(int thread_count, int queue_size, threadpool_t **ppstThreadPool);
int ThreadPool_Dispatch(threadpool_t *pstThreadPool, void (*function)(void *), void *arg);
void Threadpool_Destroy(threadpool_t *pool, threadpool_shutdown_t shutdown_mode);
#endif
2.3.2 ThrdPool.c
#include <stdio.h>
#include "ThrdPool.h"
#define THREAD_COUNT 4
#define QUEUE_SIZE 128
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 靜態初始化鎖，用於保證第16行的完整輸出
void func(void *arg)
{
static int num = 0;
pthread_mutex_lock(&mutex);
// 為方便觀察，故特意輸出該語句，並使用num來區分不同的任務
DEBUG("這是執行的第 %d 個任務", ++num);
usleep(100000); // 模擬任務耗時，100ms
pthread_mutex_unlock(&mutex);
return;
}
int main()
{
int iRet;
threadpool_t *pool;
iRet = ThreadPool_Init(THREAD_COUNT, QUEUE_SIZE, &pool);
if (iRet != E_SUCCEED)
{
return 0;
}
int i;
for (i = 0; i < 20; i++) // 生產者，向任務佇列中塞入 20 個任務
{
ThreadPool_Dispatch(pool, func, NULL);
}
usleep(500000);
// Threadpool_Destroy(pool, immediate_shutdown); // 立刻關閉執行緒池
Threadpool_Destroy(pool, graceful_shutdown); // 等任務佇列中的任務全部執行完畢再關閉
return 0;
}
2.3.3 main.c
#include <stdio.h>
#include "ThrdPool.h"
#define THREAD_COUNT 4
#define QUEUE_SIZE 128
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 靜態初始化鎖，用於保證第 15 行的完整輸出
void func(void *arg)
{
static int num = 0;
pthread_mutex_lock(&mutex);
DEBUG("這是執行的第 %d 個任務", ++num); // 為方便觀察，故特意輸出該語句，並使用num來區分不同的任務
usleep(100000); // 模擬任務耗時，100ms
pthread_mutex_unlock(&mutex);
return;
}
int main()
{
int iRet;
threadpool_t *pool;
iRet = ThreadPool_Init(THREAD_COUNT, QUEUE_SIZE, &pool);
if (iRet != E_SUCCEED)
{
return 0;
}
int i;
for (i = 0; i < 20; i++) // 生產者，向任務佇列中塞入 20 個任務
{
ThreadPool_Dispatch(pool, func, NULL);
}
usleep(500000);
// Threadpool_Destroy(pool, immediate_shutdown); // 立刻關閉執行緒池
Threadpool_Destroy(pool, graceful_shutdown); // 等任務執行完畢後方可關閉
return 0;
}
2.4 Tutorial
2.4.1 目錄結構
2.4.2 編譯、執行
