非常精简的Linux线程池实现（一）——使用互斥锁和条件变量
https://blog.csdn.net/kxcfzyk/article/details/31719687
線程池的含義跟它的名字一樣，就是一個由許多線程組成的池子。
有了線程池，在程序中使用多線程變得簡單。我們不用再自己去操心線程的創建、撤銷、管理問題，有什么要消耗大量CPU時間的任務通通直接扔到線程池里就好了，然后我們的主程序（主線程）可以繼續干自己的事去，線程池里面的線程會自動去執行這些任務。
另一方面，線程池提升了多線程程序的性能。我們不需要在大量任務需要執行時現創建大量線程，然后在任務結束時又銷毀大量線程，因為線程池里面的線程都是現成的而且能夠重復使用。一個理想的線程池能夠合理地動態調節池內線程數量，既不會因為線程過少而導致大量任務堆積，也不會因為線程過多了而增加額外的系統開銷。
線程池看上去很神奇的樣子，那它是怎么實現的呢？線程這么虛渺在的東西也能像有形的物品一樣圈在一個池子里？在只知道線程池這個名字的時候，我心里的疑惑就是這樣的。
其實線程池的原理非常簡單，它就是一個非常典型的生產者消費者同步問題。如果不知道我說的這個XXX問題也不要緊，我下面就解釋。
根據剛才描述的線程池的功能，可以看出線程池至少有兩個主要動作，一個是主程序不定時地向線程池添加任務，另一個是線程池里的線程領取任務去執行。且不論任務和執行任務是個什么概念，但是一個任務肯定只能分配給一個線程執行。
這樣就可以簡單猜想線程池的一種可能的架構了：主程序執行入隊操作，把任務添加到一個隊列里面；池子里的多個工作線程共同對這個隊列試圖執行出隊操作，這里要保證同一時刻只有一個線程出隊成功，搶奪到這個任務，其他線程繼續共同試圖出隊搶奪下一個任務。所以在實現線程池之前，我們需要一個隊列，我為這個線程池配備的隊列單獨放到了另一篇博客一個通用純C隊列的實現中。
這里的生產者就是主程序，生產任務（增加任務），消費者就是工作線程，消費任務（執行、減少任務）。因為這里涉及到多個線程同時訪問一個隊列的問題，所以我們需要互斥鎖來保護隊列，同時還需要條件變量來處理主線程通知任務到達、工作線程搶奪任務的問題。如果不熟悉條件變量，我在另一篇博客Linux C語言多線程庫Pthread中條件變量的的正確用法逐步詳解中作了詳細說明。
準備工作都差不多了，可以開始設計線程池了。一個最簡單線程池應該有什么功能呢？對于使用者來說，除了創建和銷毀線程池，最簡單的情況下只需要一個功能——添加任務。對于線程池自己來說，最簡單的情況下不需要動態調節線程數量，不需要考慮線程同步、線程死鎖等等一大堆麻煩的問題。所以最后的線程池API定義為：
?
//thread_pool.h
#ifndef THREAD_POOL_H_INCLUDED
#define THREAD_POOL_H_INCLUDED
typedef struct thread_pool *thread_pool_t;
thread_pool_t thread_pool_create(unsigned int thread_count);
void thread_pool_add_task(thread_pool_t pool, void* (*routine)(void *arg), void *arg);
void thread_pool_destroy(thread_pool_t pool);
#endif //THREAD_POOL_H_INCLUDED
創建線程池時指定線程池中應該固定包含多少工作線程，添加任務就是向線程池添加一個任務函數指針和任務函數需要的參數——這跟Pthread線程庫中的普通線程創建函數pthread_create是一樣的。根據這套線程池API，我們使用線程池的應用程序應該是這個套路：
?
?
//test.c
#include "thread_pool.h"
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
void* test(void *arg) {
int i;
for(i=0; i<5; i++) {
printf("tid:%ld task:%ld\n", pthread_self(), (long)arg);
fflush(stdout);
sleep(2);
}
return NULL;
}
int main() {
long i=0;
thread_pool_t pool;
pool=thread_pool_create(2);
for(i=0; i<5; i++) {
thread_pool_add_task(pool, test, (void*)i);
}
puts("press enter to terminate ...");
getchar();
thread_pool_destroy(pool);
return 0;
}
上面這個測試程序向線程池添加了5個相同的任務，每個任務耗時10秒，但是線程池中只有2個工作線程，所以程序的運行結果是兩個工作線程輪流把5個任務挨個做完。顯示到屏幕上就是：前10秒兩個工作線程輪流輸出自己的線程ID和當前任務的任務號0和1，各輸出5次；第二個10秒兩個工作線程輪流輸出自己的線程ID和當前任務的任務號2和3……
?
在這期間，主程序輸出“press enter to terminate ...”并等待用戶輸入，任何時候都可以按回車讓主程序繼續往下，這樣會強制終止所有工作線程并銷毀線程池，最后程序退出。test程序運行效果截圖如下：
?
最后就是線程池真正的實現了：
?
?
//thread_pool.c
#include "thread_pool.h"
#include "queue.h"
#include <stdlib.h>
#include <pthread.h>
struct thread_pool {
unsigned int thread_count;
pthread_t *threads;
queue_t tasks;
pthread_mutex_t lock;
pthread_cond_t task_ready;
};
struct task {
void* (*routine)(void *arg);
void *arg;
};
static void cleanup(pthread_mutex_t* lock) {
pthread_mutex_unlock(lock);
}
static void * worker(thread_pool_t pool) {
struct task *t;
while(1) {
pthread_mutex_lock(&pool->lock);
pthread_cleanup_push((void(*)(void*))cleanup, &pool->lock);
while(queue_isempty(pool->tasks)) {
pthread_cond_wait(&pool->task_ready, &pool->lock);
/*A condition wait (whether timed or not) is a cancellation point ... a side-effect of acting upon a cancellation request while in a condition wait is that the mutex is (in effect) re-acquired before calling the first cancellation cleanup handler.*/
}
t=(struct task*)queue_dequeue(pool->tasks);
pthread_cleanup_pop(0);
pthread_mutex_unlock(&pool->lock);
t->routine(t->arg);/*todo: report returned value*/
free(t);
}
return NULL;
}
thread_pool_t thread_pool_create(unsigned int thread_count) {
unsigned int i;
thread_pool_t pool=NULL;
pool=(thread_pool_t)malloc(sizeof(struct thread_pool));
pool->thread_count=thread_count;
pool->threads=(pthread_t*)malloc(sizeof(pthread_t)*thread_count);
pool->tasks=queue_create();
pthread_mutex_init(&pool->lock, NULL);
pthread_cond_init(&pool->task_ready, NULL);
for(i=0; i<thread_count; i++) {
pthread_create(pool->threads+i, NULL, (void*(*)(void*))worker, pool);
}
return pool;
}
void thread_pool_add_task(thread_pool_t pool, void* (*routine)(void *arg), void *arg) {
struct task *t;
pthread_mutex_lock(&pool->lock);
t=(struct task*)queue_enqueue(pool->tasks, sizeof(struct task));
t->routine=routine;
t->arg=arg;
pthread_cond_signal(&pool->task_ready);
pthread_mutex_unlock(&pool->lock);
}
void thread_pool_destroy(thread_pool_t pool) {
unsigned int i;
for(i=0; i<pool->thread_count; i++) {
pthread_cancel(pool->threads[i]);
}
for(i=0; i<pool->thread_count; i++) {
pthread_join(pool->threads[i], NULL);
}
pthread_mutex_destroy(&pool->lock);
pthread_cond_destroy(&pool->task_ready);
queue_destroy(pool->tasks);
free(pool->threads);
free(pool);
}
上面的worker函數就是工作線程函數，所有的工作線程都在執行著這個函數。它首先在互斥鎖和條件變量的保護下從任務隊列中取出一個任務，這個任務實際上是一個函數指針和調用函數所需的參數，所以執行任務就很簡單了——用任務參數調用任務函數。函數返回以后，工作線程繼續去搶任務。
?
這里沒有處理任務函數的返回值問題，理論上任務函數返回以后線程池應該用某種機制通知主程序，然后主程序獲取通過某種手段獲取返回值，但這明顯不是一個最簡單的線程池需要操心的事。實際上，應用程序可以通過全局變量或傳入的參數指針，加上額外的線程同步代碼解決返回值的通知和獲取問題。
還有一點需要注意，最后線程池銷毀時會強制終止所有處于撤銷點（cacellation points）的工作線程，如果工作線程正在任務函數中沒返回而且任務函數中有非手動創建的撤銷點，那么任務函數就會在跑到撤銷點時戛然而止，這可能導致意外結果。而如果任務函數中沒有任何線程撤銷點，那么線程池銷毀函數會一直阻塞等待直到任務函數完成后才能終止對應的工作線程并返回。
要正確處理這個問題，線程池使用者必須通過自己的線程同步代碼保證調用thread_pool_destroy之前所有任務都已經完成、終止或者取消。
?
總結
以上是生活随笔為你收集整理的非常精简的Linux线程池实现（一）——使用互斥锁和条件变量的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： LOL我不怎么会ADC 想买一个奥巴马去
- 下一篇： Linux C 实现一个简单的线程池