C 語言實現簡易線程池：任務隊列、工作線程與同步機制完整代碼

線程池核心架構

線程池包含三大組件：

• 任務隊列：線程安全的 FIFO 隊列，存儲待執行任務
• 工作線程組：固定數量的線程，循環從隊列取任務執行
• 同步機制：互斥鎖保護隊列，條件變量實現任務通知/等待

頭文件：threadpool.h

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <pthread.h>

typedef void *(*task_func_t)(void *arg);  // 任務函數類型

typedef struct thread_task {
    task_func_t func;           // 執行函數
    void *arg;                  // 函數參數
    struct thread_task *next;   // 鏈表下一節點
} thread_task_t;

typedef struct thread_pool {
    thread_task_t *head;        // 隊列頭 (哨兵節點)
    thread_task_t *tail;        // 隊列尾
    int task_count;             // 待執行任務數
    int shutdown;               // 關閉標誌
    
    pthread_mutex_t lock;       // 隊列互斥鎖
    pthread_cond_t not_empty;   // 隊列非空條件變量
    
    pthread_t *threads;         // 工作線程 ID 數組
    int thread_count;           // 線程數
} thread_pool_t;

/* 接口函數 */
thread_pool_t *thread_pool_create(int thread_count);
int thread_pool_add_task(thread_pool_t *pool, task_func_t func, void *arg);
void thread_pool_destroy(thread_pool_t *pool);

#endif

實現文件：threadpool.c

#include "threadpool.h"
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>

static void *worker_thread(void *arg) {
    thread_pool_t *pool = (thread_pool_t *)arg;
    thread_task_t *task;
    
    while (1) {
        pthread_mutex_lock(&pool->lock);
        
        /* 隊列為空且未關閉：等待 */
        while (pool->task_count == 0 && !pool->shutdown) {
            pthread_cond_wait(&pool->not_empty, &pool->lock);
        }
        
        /* 收到關閉信號且隊列空：退出 */
        if (pool->shutdown && pool->task_count == 0) {
            pthread_mutex_unlock(&pool->lock);
            pthread_exit(NULL);
        }
        
        /* 取出任務 */
        task = pool->head->next;
        pool->head->next = task->next;
        if (pool->tail == task) pool->tail = pool->head;  // 隊列變空
        pool->task_count--;
        
        pthread_mutex_unlock(&pool->lock);
        
        /* 執行任務 (鎖外執行，避免長時間持鎖) */
        task->func(task->arg);
        free(task);
    }
    return NULL;
}

thread_pool_t *thread_pool_create(int thread_count) {
    if (thread_count <= 0) thread_count = 4;
    
    thread_pool_t *pool = calloc(1, sizeof(thread_pool_t));
    if (!pool) return NULL;
    
    /* 初始化哨兵節點 */
    pool->head = pool->tail = calloc(1, sizeof(thread_task_t));
    if (!pool->head) { free(pool); return NULL; }
    
    pool->thread_count = thread_count;
    pool->threads = calloc(thread_count, sizeof(pthread_t));
    if (!pool->threads) { free(pool->head); free(pool); return NULL; }
    
    pthread_mutex_init(&pool->lock, NULL);
    pthread_cond_init(&pool->not_empty, NULL);
    
    /* 創建工作線程 */
    for (int i = 0; i < thread_count; i++) {
        if (pthread_create(&pool->threads[i], NULL, worker_thread, pool) != 0) {
            pool->shutdown = 1;
            pthread_cond_broadcast(&pool->not_empty);
            for (int j = 0; j < i; j++) pthread_join(pool->threads[j], NULL);
            pthread_mutex_destroy(&pool->lock);
            pthread_cond_destroy(&pool->not_empty);
            free(pool->threads);
            free(pool->head);
            free(pool);
            return NULL;
        }
    }
    return pool;
}

int thread_pool_add_task(thread_pool_t *pool, task_func_t func, void *arg) {
    if (!pool || !func) return -1;
    
    thread_task_t *task = calloc(1, sizeof(thread_task_t));
    if (!task) return -1;
    task->func = func;
    task->arg = arg;
    
    pthread_mutex_lock(&pool->lock);
    if (pool->shutdown) {
        pthread_mutex_unlock(&pool->lock);
        free(task);
        return -1;
    }
    
    pool->tail->next = task;
    pool->tail = task;
    pool->task_count++;
    
    pthread_cond_signal(&pool->not_empty);
    pthread_mutex_unlock(&pool->lock);
    return 0;
}

void thread_pool_destroy(thread_pool_t *pool) {
    if (!pool) return;
    
    pthread_mutex_lock(&pool->lock);
    pool->shutdown = 1;
    pthread_cond_broadcast(&pool->not_empty);  // 喚醒所有等待線程
    pthread_mutex_unlock(&pool->lock);
    
    /* 等待所有線程退出 */
    for (int i = 0; i < pool->thread_count; i++) {
        pthread_join(pool->threads[i], NULL);
    }
    
    /* 清理剩餘任務 */
    thread_task_t *curr = pool->head->next;
    while (curr) {
        thread_task_t *next = curr->next;
        free(curr);
        curr = next;
    }
    
    pthread_mutex_destroy(&pool->lock);
    pthread_cond_destroy(&pool->not_empty);
    free(pool->threads);
    free(pool->head);
    free(pool);
}

測試程序：main.c

#include "threadpool.h"
#include <stdio.h>
#include <unistd.h>

void *task1(void *arg) {
    printf("[任務1] 開始執行 (線程 %lu)\n", (unsigned long)pthread_self());
    sleep(2);
    printf("[任務1] 執行結束\n");
    return NULL;
}

void *task2(void *arg) {
    int id = *(int *)arg;
    printf("[任務2-%d] 開始執行\n", id);
    sleep(1);
    printf("[任務2-%d] 執行結束\n", id);
    return NULL;
}

void *task3(void *arg) {
    printf("[任務3] 快速任務\n");
    return NULL;
}

int main() {
    thread_pool_t *pool = thread_pool_create(3);  // 3 個工作線程
    if (!pool) { fprintf(stderr, "線程池創建失敗\n"); return 1; }
    
    int ids[5] = {1,2,3,4,5};
    
    /* 提交任務 */
    thread_pool_add_task(pool, task1, NULL);
    for (int i = 0; i < 5; i++) {
        thread_pool_add_task(pool, task2, &ids[i]);
    }
    thread_pool_add_task(pool, task3, NULL);
    
    printf("主線程: 所有任務已提交，等待完成...\n");
    sleep(5);  // 簡單等待，實際應用可用計數器/條件變量同步
    
    thread_pool_destroy(pool);
    printf("主線程: 線程池已銷毀\n");
    return 0;
}

編譯與運行

gcc -Wall -O2 -pthread -o threadpool_demo threadpool.c main.c
./threadpool_demo

關鍵設計細節解析

生產級擴展建議

• 動態伸縮：根據隊列長度增減線程，設置核心/最大線程數
• 任務優先級：優先級隊列 (堆/多級隊列)，高優先級任務插隊
• 定時/周期任務：整合 timerfd 或單獨調度線程
• Future/Promise：任務返回值獲取，支持同步等待結果
• 工作盜演算法：多隊列 + work-stealing，減少鎖競爭 (如 Java ForkJoinPool)
• 內存池：任務節點預分配，避免頻繁 malloc/free
• 監控統計：完成任務數、隊列長度、線程利用率導出