[[442559]]

1. 線程池原理

我們使用線程的時(shí)候就去創(chuàng)建一個(gè)線程，這樣實(shí)現(xiàn)起來(lái)非常簡(jiǎn)便，但是就會(huì)有一個(gè)問(wèn)題：如果并發(fā)的線程數(shù)量很多，并且每個(gè)線程都是執(zhí)行一個(gè)時(shí)間很短的任務(wù)就結(jié)束了，這樣頻繁創(chuàng)建線程就會(huì)大大降低系統(tǒng)的效率，因?yàn)轭l繁創(chuàng)建線程和銷毀線程需要時(shí)間。

那么有沒(méi)有一種辦法使得線程可以復(fù)用，就是執(zhí)行完一個(gè)任務(wù)，并不被銷毀，而是可以繼續(xù)執(zhí)行其他的任務(wù)呢?

線程池是一種多線程處理形式，處理過(guò)程中將任務(wù)添加到隊(duì)列，然后在創(chuàng)建線程后自動(dòng)啟動(dòng)這些任務(wù)。線程池線程都是后臺(tái)線程。每個(gè)線程都使用默認(rèn)的堆棧大小，以默認(rèn)的優(yōu)先級(jí)運(yùn)行，并處于多線程單元中。

如果某個(gè)線程在托管代碼中空閑(如正在等待某個(gè)事件), 則線程池將插入另一個(gè)輔助線程來(lái)使所有處理器保持繁忙。如果所有線程池線程都始終保持繁忙，但隊(duì)列中包含掛起的工作，則線程池將在一段時(shí)間后創(chuàng)建另一個(gè)輔助線程但線程的數(shù)目永遠(yuǎn)不會(huì)超過(guò)最大值。超過(guò)最大值的線程可以排隊(duì)，但他們要等到其他線程完成后才啟動(dòng)。

在各個(gè)編程語(yǔ)言的語(yǔ)種中都有線程池的概念，并且很多語(yǔ)言中直接提供了線程池，作為程序猿直接使用就可以了，下面給大家介紹一下線程池的實(shí)現(xiàn)原理：

線程池的組成主要分為 3 個(gè)部分，這三部分配合工作就可以得到一個(gè)完整的線程池：

1）.任務(wù)隊(duì)列，存儲(chǔ)需要處理的任務(wù)，由工作的線程來(lái)處理這些任務(wù)

• 通過(guò)線程池提供的 API 函數(shù)，將一個(gè)待處理的任務(wù)添加到任務(wù)隊(duì)列，或者從任務(wù)隊(duì)列中刪除
• 已處理的任務(wù)會(huì)被從任務(wù)隊(duì)列中刪除
• 線程池的使用者，也就是調(diào)用線程池函數(shù)往任務(wù)隊(duì)列中添加任務(wù)的線程就是生產(chǎn)者線程

2）.工作的線程(任務(wù)隊(duì)列任務(wù)的消費(fèi)者) ，N個(gè)

• 線程池中維護(hù)了一定數(shù)量的工作線程，他們的作用是是不停的讀任務(wù)隊(duì)列，從里邊取出任務(wù)并處理
• 工作的線程相當(dāng)于是任務(wù)隊(duì)列的消費(fèi)者角色，
• 如果任務(wù)隊(duì)列為空，工作的線程將會(huì)被阻塞 (使用條件變量 / 信號(hào)量阻塞)
• 如果阻塞之后有了新的任務(wù)，由生產(chǎn)者將阻塞解除，工作線程開(kāi)始工作

3）.管理者線程(不處理任務(wù)隊(duì)列中的任務(wù))，1個(gè)

• 它的任務(wù)是周期性的對(duì)任務(wù)隊(duì)列中的任務(wù)數(shù)量以及處于忙狀態(tài)的工作線程個(gè)數(shù)進(jìn)行檢測(cè)
• 當(dāng)任務(wù)過(guò)多的時(shí)候，可以適當(dāng)?shù)膭?chuàng)建一些新的工作線程
• 當(dāng)任務(wù)過(guò)少的時(shí)候，可以適當(dāng)?shù)匿N毀一些工作的線程

2. 任務(wù)隊(duì)列

// 任務(wù)結(jié)構(gòu)體 
typedef struct Task 
{ 
    void (*function)(void* arg); 
    void* arg; 
}Task; 

3. 線程池定義

// 線程池結(jié)構(gòu)體 
struct ThreadPool 
{ 
    // 任務(wù)隊(duì)列 
    Task* taskQ; 
    int queueCapacity;  // 容量 
    int queueSize;      // 當(dāng)前任務(wù)個(gè)數(shù) 
    int queueFront;     // 隊(duì)頭 -> 取數(shù)據(jù) 
    int queueRear;      // 隊(duì)尾 -> 放數(shù)據(jù) 
 
    pthread_t managerID;    // 管理者線程ID 
    pthread_t *threadIDs;   // 工作的線程ID 
    int minNum;             // 最小線程數(shù)量 
    int maxNum;             // 最大線程數(shù)量 
    int busyNum;            // 忙的線程的個(gè)數(shù) 
    int liveNum;            // 存活的線程的個(gè)數(shù) 
    int exitNum;            // 要銷毀的線程個(gè)數(shù) 
    pthread_mutex_t mutexPool;  // 鎖整個(gè)的線程池 
    pthread_mutex_t mutexBusy;  // 鎖busyNum變量 
    pthread_cond_t notFull;     // 任務(wù)隊(duì)列是不是滿了 
    pthread_cond_t notEmpty;    // 任務(wù)隊(duì)列是不是空了 
 
    int shutdown;           // 是不是要銷毀線程池, 銷毀為1, 不銷毀為0 
}; 

4. 頭文件聲明

#ifndef _THREADPOOL_H 
#define _THREADPOOL_H 
 
typedef struct ThreadPool ThreadPool; 
// 創(chuàng)建線程池并初始化 
ThreadPool *threadPoolCreate(int min, int max, int queueSize); 
 
// 銷毀線程池 
int threadPoolDestroy(ThreadPool* pool); 
 
// 給線程池添加任務(wù) 
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg); 
 
// 獲取線程池中工作的線程的個(gè)數(shù) 
int threadPoolBusyNum(ThreadPool* pool); 
 
// 獲取線程池中活著的線程的個(gè)數(shù) 
int threadPoolAliveNum(ThreadPool* pool); 
 
////////////////////// 
// 工作的線程(消費(fèi)者線程)任務(wù)函數(shù) 
void* worker(void* arg); 
// 管理者線程任務(wù)函數(shù) 
void* manager(void* arg); 
// 單個(gè)線程退出 
void threadExit(ThreadPool* pool); 
#endif  // _THREADPOOL_H 

5. 源文件定義

ThreadPool* threadPoolCreate(int min, int max, int queueSize) 
{ 
    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool)); 
    do  
    { 
        if (pool == NULL) 
        { 
            printf("malloc threadpool fail...\n"); 
            break; 
        } 
 
        pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max); 
        if (pool->threadIDs == NULL) 
        { 
            printf("malloc threadIDs fail...\n"); 
            break; 
        } 
        memset(pool->threadIDs, 0, sizeof(pthread_t) * max); 
        pool->minNum = min; 
        pool->maxNum = max; 
        pool->busyNum = 0; 
        pool->liveNum = min;    // 和最小個(gè)數(shù)相等 
        pool->exitNum = 0; 
 
        if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 || 
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 || 
            pthread_cond_init(&pool->notEmpty, NULL) != 0 || 
            pthread_cond_init(&pool->notFull, NULL) != 0) 
        { 
            printf("mutex or condition init fail...\n"); 
            break; 
        } 
 
        // 任務(wù)隊(duì)列 
        pool->taskQ = (Task*)malloc(sizeof(Task) * queueSize); 
        pool->queueCapacity = queueSize; 
        pool->queueSize = 0; 
        pool->queueFront = 0; 
        pool->queueRear = 0; 
 
        pool->shutdown = 0; 
 
        // 創(chuàng)建線程 
        pthread_create(&pool->managerID, NULL, manager, pool); 
        for (int i = 0; i < min; ++i) 
        { 
            pthread_create(&pool->threadIDs[i], NULL, worker, pool); 
        } 
        return pool; 
    } while (0); 
 
    // 釋放資源 
    if (pool && pool->threadIDs) free(pool->threadIDs); 
    if (pool && pool->taskQ) free(pool->taskQ); 
    if (pool) free(pool); 
 
    return NULL; 
} 
 
int threadPoolDestroy(ThreadPool* pool) 
{ 
    if (pool == NULL) 
    { 
        return -1; 
    } 
 
    // 關(guān)閉線程池 
    pool->shutdown = 1; 
    // 阻塞回收管理者線程 
    pthread_join(pool->managerID, NULL); 
    // 喚醒阻塞的消費(fèi)者線程 
    for (int i = 0; i < pool->liveNum; ++i) 
    { 
        pthread_cond_signal(&pool->notEmpty); 
    } 
    // 釋放堆內(nèi)存 
    if (pool->taskQ) 
    { 
        free(pool->taskQ); 
    } 
    if (pool->threadIDs) 
    { 
        free(pool->threadIDs); 
    } 
 
    pthread_mutex_destroy(&pool->mutexPool); 
    pthread_mutex_destroy(&pool->mutexBusy); 
    pthread_cond_destroy(&pool->notEmpty); 
    pthread_cond_destroy(&pool->notFull); 
 
    free(pool); 
    pool = NULL; 
 
    return 0; 
} 
 
 
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg) 
{ 
    pthread_mutex_lock(&pool->mutexPool); 
    while (pool->queueSize == pool->queueCapacity && !pool->shutdown) 
    { 
        // 阻塞生產(chǎn)者線程 
        pthread_cond_wait(&pool->notFull, &pool->mutexPool); 
    } 
    if (pool->shutdown) 
    { 
        pthread_mutex_unlock(&pool->mutexPool); 
        return; 
    } 
    // 添加任務(wù) 
    pool->taskQ[pool->queueRear].function = func; 
    pool->taskQ[pool->queueRear].arg = arg; 
    pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity; 
    pool->queueSize++; 
 
    pthread_cond_signal(&pool->notEmpty); 
    pthread_mutex_unlock(&pool->mutexPool); 
} 
 
int threadPoolBusyNum(ThreadPool* pool) 
{ 
    pthread_mutex_lock(&pool->mutexBusy); 
    int busyNum = pool->busyNum; 
    pthread_mutex_unlock(&pool->mutexBusy); 
    return busyNum; 
} 
 
int threadPoolAliveNum(ThreadPool* pool) 
{ 
    pthread_mutex_lock(&pool->mutexPool); 
    int aliveNum = pool->liveNum; 
    pthread_mutex_unlock(&pool->mutexPool); 
    return aliveNum; 
} 
 
void* worker(void* arg) 
{ 
    ThreadPool* pool = (ThreadPool*)arg; 
 
    while (1) 
    { 
        pthread_mutex_lock(&pool->mutexPool); 
        // 當(dāng)前任務(wù)隊(duì)列是否為空 
        while (pool->queueSize == 0 && !pool->shutdown) 
        { 
            // 阻塞工作線程 
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool); 
 
            // 判斷是不是要銷毀線程 
            if (pool->exitNum > 0) 
            { 
                pool->exitNum--; 
                if (pool->liveNum > pool->minNum) 
                { 
                    pool->liveNum--; 
                    pthread_mutex_unlock(&pool->mutexPool); 
                    threadExit(pool); 
                } 
            } 
        } 
 
        // 判斷線程池是否被關(guān)閉了 
        if (pool->shutdown) 
        { 
            pthread_mutex_unlock(&pool->mutexPool); 
            threadExit(pool); 
        } 
 
        // 從任務(wù)隊(duì)列中取出一個(gè)任務(wù) 
        Task task; 
        task.function = pool->taskQ[pool->queueFront].function; 
        task.arg = pool->taskQ[pool->queueFront].arg; 
        // 移動(dòng)頭結(jié)點(diǎn) 
        pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity; 
        pool->queueSize--; 
        // 解鎖 
        pthread_cond_signal(&pool->notFull); 
        pthread_mutex_unlock(&pool->mutexPool); 
 
        printf("thread %ld start working...\n", pthread_self()); 
        pthread_mutex_lock(&pool->mutexBusy); 
        pool->busyNum++; 
        pthread_mutex_unlock(&pool->mutexBusy); 
        task.function(task.arg); 
        free(task.arg); 
        task.arg = NULL; 
 
        printf("thread %ld end working...\n", pthread_self()); 
        pthread_mutex_lock(&pool->mutexBusy); 
        pool->busyNum--; 
        pthread_mutex_unlock(&pool->mutexBusy); 
    } 
    return NULL; 
} 
 
void* manager(void* arg) 
{ 
    ThreadPool* pool = (ThreadPool*)arg; 
    while (!pool->shutdown) 
    { 
        // 每隔3s檢測(cè)一次 
        sleep(3); 
 
        // 取出線程池中任務(wù)的數(shù)量和當(dāng)前線程的數(shù)量 
        pthread_mutex_lock(&pool->mutexPool); 
        int queueSize = pool->queueSize; 
        int liveNum = pool->liveNum; 
        pthread_mutex_unlock(&pool->mutexPool); 
 
        // 取出忙的線程的數(shù)量 
        pthread_mutex_lock(&pool->mutexBusy); 
        int busyNum = pool->busyNum; 
        pthread_mutex_unlock(&pool->mutexBusy); 
 
        // 添加線程 
        // 任務(wù)的個(gè)數(shù)>存活的線程個(gè)數(shù) && 存活的線程數(shù)<最大線程數(shù) 
        if (queueSize > liveNum && liveNum < pool->maxNum) 
        { 
            pthread_mutex_lock(&pool->mutexPool); 
            int counter = 0; 
            for (int i = 0; i < pool->maxNum && counter < NUMBER 
                && pool->liveNum < pool->maxNum; ++i) 
            { 
                if (pool->threadIDs[i] == 0) 
                { 
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool); 
                    counter++; 
                    pool->liveNum++; 
                } 
            } 
            pthread_mutex_unlock(&pool->mutexPool); 
        } 
        // 銷毀線程 
        // 忙的線程*2 < 存活的線程數(shù) && 存活的線程>最小線程數(shù) 
        if (busyNum * 2 < liveNum && liveNum > pool->minNum) 
        { 
            pthread_mutex_lock(&pool->mutexPool); 
            pool->exitNum = NUMBER; 
            pthread_mutex_unlock(&pool->mutexPool); 
            // 讓工作的線程自殺 
            for (int i = 0; i < NUMBER; ++i) 
            { 
                pthread_cond_signal(&pool->notEmpty); 
            } 
        } 
    } 
    return NULL; 
} 
 
void threadExit(ThreadPool* pool) 
{ 
    pthread_t tid = pthread_self(); 
    for (int i = 0; i < pool->maxNum; ++i) 
    { 
        if (pool->threadIDs[i] == tid) 
        { 
            pool->threadIDs[i] = 0; 
            printf("threadExit() called, %ld exiting...\n", tid); 
            break; 
        } 
    } 
    pthread_exit(NULL); 
} 

6. 測(cè)試代碼

void taskFunc(void* arg) 
{ 
    int num = *(int*)arg; 
    printf("thread %ld is working, number = %d\n", 
        pthread_self(), num); 
    sleep(1); 
} 
 
int main() 
{ 
    // 創(chuàng)建線程池 
    ThreadPool* pool = threadPoolCreate(3, 10, 100); 
    for (int i = 0; i < 100; ++i) 
    { 
        int* num = (int*)malloc(sizeof(int)); 
        *num = i + 100; 
        threadPoolAdd(pool, taskFunc, num); 
    } 
 
    sleep(30); 
 
    threadPoolDestroy(pool); 
    return 0; 
}