您现在的位置：程式碼之家 >> 編程語言 > >> C語言 >> C >> 關於C

C 語言實作簡易執行緒池：完整原始碼與設計解析

編輯：關於C

設計目標

固定大小的工作者執行緒池
任務佇列（FIFO）支援動態新增任務
互斥鎖 + 條件變數同步化
優雅關閉：完成所有任務後回收資源

核心資料結構

typedef struct task {
    void *(*func)(void *arg); // 任務函式指標
    void *arg;                // 參數
    struct task *next;        // 串列下一個
} task_t;

typedef struct thread_pool {
    task_t *head, *tail;      // 任務佇列頭尾
    int queue_size;           // 待處理任務數
    pthread_t *threads;       // 工作者執行緒陣列
    int thread_num;           // 執行緒總數
    int idle_num;             // 閒置執行緒數
    int shutdown;             // 關閉旗標
    pthread_mutex_t lock;     // 保護佇列
    pthread_cond_t cond;      // 任務到達/關閉通知
} thread_pool_t;

關鍵函式實作

初始化 `thread_pool_init()`

int thread_pool_init(int size) {
    pool = calloc(1, sizeof(thread_pool_t));
    pool->threads = calloc(size, sizeof(pthread_t));
    pool->thread_num = pool->idle_num = size;
    pool->shutdown = 0;
    pthread_mutex_init(&pool->lock, NULL);
    pthread_cond_init(&pool->cond, NULL);
    for (int i = 0; i < size; ++i)
        pthread_create(&pool->threads[i], NULL, worker, (void *)(intptr_t)i);
    return 0;
}

工作者主迴圈 `worker()`

void *worker(void *arg) {
    int id = (intptr_t)arg;
    printf("Worker %d started\n", id);
    while (1) {
        pthread_mutex_lock(&pool->lock);
        while (pool->queue_size == 0 && !pool->shutdown)
            pthread_cond_wait(&pool->cond, &pool->lock); // 等待任務
        if (pool->shutdown) {
            pthread_mutex_unlock(&pool->lock);
            printf("Worker %d exit\n", id);
            pthread_exit(NULL);
        }
        // 取出任務
        task_t *t = pool->head;
        pool->head = t->next;
        if (!pool->head) pool->tail = NULL;
        --pool->queue_size;
        --pool->idle_num;
        pthread_mutex_unlock(&pool->lock);

        t->func(t->arg); // 執行任務
        free(t);

        pthread_mutex_lock(&pool->lock);
        ++pool->idle_num;
        pthread_mutex_unlock(&pool->lock);
    }
}

新增任務 `thread_pool_add_task()`

int thread_pool_add_task(void *(*func)(void *), void *arg) {
    task_t *t = malloc(sizeof(task_t));
    t->func = func; t->arg = arg; t->next = NULL;
    pthread_mutex_lock(&pool->lock);
    if (!pool->head) pool->head = pool->tail = t;
    else { pool->tail->next = t; pool->tail = t; }
    ++pool->queue_size;
    pthread_cond_signal(&pool->cond); // 喚醒一個工作者
    pthread_mutex_unlock(&pool->lock);
    return 0;
}

銷毀池 `thread_pool_destroy()`

int thread_pool_destroy(void) {
    pthread_mutex_lock(&pool->lock);
    if (pool->shutdown) { pthread_mutex_unlock(&pool->lock); return -1; }
    pool->shutdown = 1;
    pthread_cond_broadcast(&pool->cond); // 喚醒所有
    pthread_mutex_unlock(&pool->lock);

    for (int i = 0; i < pool->thread_num; ++i)
        pthread_join(pool->threads[i], NULL);

    // 釋放剩餘任務、鎖、條件變數、池本身
    // ... 略 ...
    return 0;
}

測試程式

void *demo_task(void *arg) {
    int id = (intptr_t)arg;
    printf("Task %d running on thread %lu\n", id, pthread_self());
    usleep(500000);
    return NULL;
}

int main(void) {
    thread_pool_init(4);
    for (int i = 0; i < 10; ++i)
        thread_pool_add_task(demo_task, (void *)(intptr_t)i);
    sleep(2);
    thread_pool_destroy();
    return 0;
}