本文共 5428 字，大约阅读时间需要 18 分钟。

?????????

??????????

????????????????????????????????????????????????????

??????????

?????????????????????

1. ?????

??????????????

• ?????????????????
• ????????????????????
• ???????????????????????

2. ???????

??????????????????????

• ??ID???????????????
• ??????????????????
• ???????????????????
• ??????????????????

3. ???????

??????? _threadPool ?????????

• ???????????????????????????
• ????????????????????????????
• ?????????????????
• ???????????????????

4. ????????

??????????????

????????

1. ??????????

?????? pthread_create ??????? thread_callback ?????????????????????????????????????

2. ????????

?????????????????????????????????????????????? pthread_cond_signal ?????????????????

3. ????????

????????????????????????? pthread_cond_broadcast ???????????????????

??????????

??????????????

• ?????????????????????????
• ????????????????????????????????
• ?????????????????????????

??????????

??????????????

#include 
   
    
#include 
    
     
#include 
     
      
#include 
      
       
#include 
       
        
#include 
        
          #define LL_ADD(item, list) do { item->prev = NULL; item->next = list; if (list != NULL) list->prev = item; list = item; } while (0) #define LL_REMOVE(item, list) do { if (item->prev != NULL) item->prev->next = item->next; if (item->next != NULL) item->next->prev = item->prev; if (list == item) list = item->next; item->prev = item->next = NULL; } while (0) typedef void* (*callback)(void* arg); struct node_task { callback func; void* user_data; struct node_task* prev; struct node_task* next; }; struct node_worker { pthread_t tid; int terminate; struct _threadPool* pool; struct node_worker* prev; struct node_worker* next; }; struct _threadPool { pthread_mutex_t mtx; pthread_cond_t cond; struct node_task* tasks; struct node_worker* workers; }; typedef struct _threadPool nThreadPool; void* thread_callback(void* arg) { if (arg == NULL) { perror("Illegal parameters, and exit worker thread\n"); pthread_exit(NULL); } struct node_worker* worker = (struct node_worker*)arg; nThreadPool* pool = worker->pool; if (pool == NULL) { perror("Illegal poll, and exit worker thread\n"); pthread_exit(NULL); } while (1) { pthread_mutex_lock(&pool->mtx); while (pool->tasks == NULL && !worker->terminate) { pthread_cond_wait(&pool->cond, &pool->mtx); } if (worker->terminate) { pthread_mutex_unlock(&pool->mtx); break; } struct node_task* task = pool->tasks; if (task != NULL) { LL_REMOVE(task, pool->tasks); } pthread_mutex_unlock(&pool->mtx); if (task != NULL) { task->func(task->user_data); } free(task); } free(worker); printf("thread ID: %lu exit!\n", pthread_self()); pthread_exit(NULL); } int nThreadPoolCreate(nThreadPool* pool, int numWorkers) { if (pool == NULL || numWorkers < 0) return -1; memset(pool, 0, sizeof(nThreadPool)); pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER; memcpy(&pool->mtx, &blank_mutex, sizeof(pthread_mutex_t)); pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER; memcpy(&pool->cond, &blank_cond, sizeof(pthread_cond_t)); int i = 0; for (; i < numWorkers; i++) { struct node_worker* worker = (struct node_worker*)malloc(sizeof(struct node_worker)); if (worker == NULL) { perror("malloc"); return -1; } memset(worker, 0, sizeof(struct node_worker)); worker->pool = pool; int ret = pthread_create(&worker->tid, NULL, thread_callback, worker); if (ret) { perror("pthread_create"); free(worker); return -2; } LL_ADD(worker, pool->workers); } return 0; } int nThreadPoolDestory(nThreadPool* pool) { struct node_worker* worker = NULL; for (worker = pool->workers; worker != NULL; worker = worker->next) { worker->terminate = 1; } pthread_mutex_lock(&pool->mtx); pool->workers = NULL; struct node_task* task = NULL; for (task = pool->tasks; task != NULL; task = task->next) { free(task); } pool->tasks = NULL; pthread_cond_broadcast(&pool->cond); pthread_mutex_unlock(&pool->mtx); return 0; } int nThreadPoolPushTask(nThreadPool* pool, struct node_task* task) { pthread_mutex_lock(&pool->mtx); LL_ADD(task, pool->tasks); pthread_cond_signal(&pool->cond); pthread_mutex_unlock(&pool->mtx); return 0; }
        
       
      
     
    
   

????

??????????????????????????????????????????????

#include 
   
    
#include 
    
     
#include 
     
      
#include 
      
       
#define MAX_THREADS 10
#define COUNTER_SIZE 1000
void counter(struct node_task* task) {
    int index = *(int*)task->user_data;
    printf("index : %d, selfid : %lu\n", index, pthread_self());
    free(task->user_data);
}
int main(int argc, char* argv[]) {
    nThreadPool pool;
    nThreadPoolCreate(&pool, MAX_THREADS);
    int i = 0;
    for (; i < COUNTER_SIZE; i++) {
        struct node_task* task = (struct node_task*)malloc(sizeof(struct node_task));
        if (task == NULL) {
            perror("malloc");
            exit(EXIT_FAILURE);
        }
        task->func = (callback)counter;
        task->user_data = malloc(sizeof(int));
        *(int*)task->user_data = i;
        nThreadPoolPushTask(&pool, task);
    }
    printf("Press any key to destory the thread pool\n");
    nThreadPoolDestory(&pool);
    getchar();
    exit(EXIT_SUCCESS);
}
      
     
    
   

??????????????????????????