日本黄色一级经典视频|伊人久久精品视频|亚洲黄色色周成人视频九九九|av免费网址黄色小短片|黄色Av无码亚洲成年人|亚洲1区2区3区无码|真人黄片免费观看|无码一级小说欧美日免费三级|日韩中文字幕91在线看|精品久久久无码中文字幕边打电话

當(dāng)前位置：首頁 > 單片機(jī) > 程序喵大人

手寫線程池 - C語言版

時間：2025-12-07 19:16:43
手機(jī)看文章

掃描二維碼
隨時隨地手機(jī)看文章

1. 線程池原理

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

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

線程池是一種多線程處理形式，處理過程中將任務(wù)添加到隊列，然后在創(chuàng)建線程后自動啟動這些任務(wù)。線程池線程都是后臺線程。每個線程都使用默認(rèn)的堆棧大小，以默認(rèn)的優(yōu)先級運(yùn)行，并處于多線程單元中。如果某個線程在托管代碼中空閑（如正在等待某個事件）, 則線程池將插入另一個輔助線程來使所有處理器保持繁忙。如果所有線程池線程都始終保持繁忙，但隊列中包含掛起的工作，則線程池將在一段時間后創(chuàng)建另一個輔助線程但線程的數(shù)目永遠(yuǎn)不會超過最大值。超過最大值的線程可以排隊，但他們要等到其他線程完成后才啟動。

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

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

  1. 任務(wù)隊列，存儲需要處理的任務(wù)，由工作的線程來處理這些任務(wù)
  • 通過線程池提供的 API 函數(shù)，將一個待處理的任務(wù)添加到任務(wù)隊列，或者從任務(wù)隊列中刪除
  • 已處理的任務(wù)會被從任務(wù)隊列中刪除
  • 線程池的使用者，也就是調(diào)用線程池函數(shù)往任務(wù)隊列中添加任務(wù)的線程就是生產(chǎn)者線程
  1. 工作的線程（任務(wù)隊列任務(wù)的消費(fèi)者） ，N個
  • 線程池中維護(hù)了一定數(shù)量的工作線程，他們的作用是是不停的讀任務(wù)隊列，從里邊取出任務(wù)并處理
  • 工作的線程相當(dāng)于是任務(wù)隊列的消費(fèi)者角色，
  • 如果任務(wù)隊列為空，工作的線程將會被阻塞 (使用條件變量 / 信號量阻塞)
  • 如果阻塞之后有了新的任務(wù)，由生產(chǎn)者將阻塞解除，工作線程開始工作
  1. 管理者線程（不處理任務(wù)隊列中的任務(wù)），1個
  • 它的任務(wù)是周期性的對任務(wù)隊列中的任務(wù)數(shù)量以及處于忙狀態(tài)的工作線程個數(shù)進(jìn)行檢測
  • 當(dāng)任務(wù)過多的時候，可以適當(dāng)?shù)膭?chuàng)建一些新的工作線程
  • 當(dāng)任務(wù)過少的時候，可以適當(dāng)?shù)匿N毀一些工作的線程

2. 任務(wù)隊列 

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

3. 線程池定義 

// 線程池結(jié)構(gòu)體
struct ThreadPool
{
 // 任務(wù)隊列
 Task* taskQ;
 int queueCapacity;  // 容量
 int queueSize;      // 當(dāng)前任務(wù)個數(shù)
 int queueFront;     // 隊頭 -> 取數(shù)據(jù)
 int queueRear;      // 隊尾 -> 放數(shù)據(jù)

 pthread_t managerID;    // 管理者線程ID
 pthread_t *threadIDs;   // 工作的線程ID
 int minNum;             // 最小線程數(shù)量
 int maxNum;             // 最大線程數(shù)量
 int busyNum;            // 忙的線程的個數(shù)
 int liveNum;            // 存活的線程的個數(shù)
 int exitNum;            // 要銷毀的線程個數(shù)
 pthread_mutex_t mutexPool;  // 鎖整個的線程池
 pthread_mutex_t mutexBusy;  // 鎖busyNum變量
 pthread_cond_t notFull;     // 任務(wù)隊列是不是滿了
 pthread_cond_t notEmpty;    // 任務(wù)隊列是不是空了

 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);

// 獲取線程池中工作的線程的個數(shù)
int threadPoolBusyNum(ThreadPool* pool);

// 獲取線程池中活著的線程的個數(shù)
int threadPoolAliveNum(ThreadPool* pool);

//////////////////////
// 工作的線程(消費(fèi)者線程)任務(wù)函數(shù)
void* worker(void* arg);
// 管理者線程任務(wù)函數(shù)
void* manager(void* arg);
// 單個線程退出
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;    // 和最小個數(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ù)隊列
 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ù)隊列是否為空 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ù)隊列中取出一個任務(wù)
 Task task;
 task.function = pool->taskQ[pool->queueFront].function;
 task.arg = pool->taskQ[pool->queueFront].arg;
 // 移動頭結(jié)點
 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檢測一次
 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ù)的個數(shù)>存活的線程個數(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. 測試代碼 

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;
}
本站聲明： 本文章由作者或相關(guān)機(jī)構(gòu)授權(quán)發(fā)布，目的在于傳遞更多信息，并不代表本站贊同其觀點，本站亦不保證或承諾內(nèi)容真實性等。需要轉(zhuǎn)載請聯(lián)系該專欄作者，如若文章內(nèi)容侵犯您的權(quán)益，請及時聯(lián)系本站刪除。
關(guān)閉