Linux 線程池實戰：條件變量與互斥鎖構建生產者-消費者模型

專案背景

本文展示一個基於 POSIX 線程（pthread）的線程池範例：多個計算線程 產生數列項（1/i），兩個消費者線程 分別執行「累加」與「交替加減」聚合，透過條件變量 + 互斥鎖同步，其中一個消費者採用超時等待機制。程式可直接編譯執行，輸出完整計算過程。

核心資料結構

#define MAXS 1000
#define MAXTDS 5  // 線程池大小

double myjg[MAXS + 1]; // myjg[0] 存當前已計算索引 i，myjg[1..i] 存 1/i
int max;               // 用戶輸入 N

pthread_mutex_t eventlock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t myevent = PTHREAD_COND_INITIALIZER;
pthread_t threads[MAXTDS + 2]; // 5 計算 + 2 消費者
int isend = 0;                 // 計算結束旗標

三類線程職責

1. 計算線程 mycomp（5 個）

• 從共享陣列取得下一個待計算索引 i。
• 計算 1.0 / i 存入 myjg[i]，更新 myjg[0]。
• 每 MAXTDS 次或最後一項時，pthread_cond_broadcast 喚醒消費者。
• 最後一項完成後設 isend=1 並 signal 結束。

2. 累加消費者 myprint1（超時等待）

void *myprint1(void *arg) {
    int cur = 1;
    double sum = 0;
    struct timeval now;
    struct timespec timeout;

    while (cur <= max) {
        gettimeofday(&now);
        timeout.tv_sec = now.tv_sec + 1;      // 1 秒超時
        timeout.tv_nsec = now.tv_usec * 1000;

        pthread_mutex_lock(&eventlock);
        int rc = pthread_cond_timedwait(&myevent, &eventlock, &timeout);
        if (rc == 0) {
            int ready = (int)myjg[0];
            pthread_mutex_unlock(&eventlock);
            for (; cur <= ready; cur++) {
                sum += myjg[cur];
                printf("1/%d: %.10f added result %.10f\n", cur, myjg[cur], sum);
            }
        } else if (rc == ETIMEDOUT) {
            printf("time out\n");
            pthread_mutex_unlock(&eventlock);
        } else {
            pthread_mutex_unlock(&eventlock);
        }
    }
    printf("================add finish!============ result: %.10f\n", sum);
    return NULL;
}

超時機制意義：避免永久阻塞，定期輪詢可實現心跳、進度顯示、優雅取消。

3. 交替加減消費者 myprint2（無限等待）

void *myprint2(void *arg) {
    int cur = 1;
    double sum = 0;
    int sign = 1;

    while (cur <= max) {
        sleep(2); // 模擬較慢處理
        pthread_mutex_lock(&eventlock);
        int rc = pthread_cond_wait(&myevent, &eventlock); // 永久等待
        if (rc == 0) {
            int ready = (int)myjg[0];
            pthread_mutex_unlock(&eventlock);
            for (; cur <= ready; cur++) {
                sum += sign * myjg[cur];
                printf("*******1/%d: %.10f computed result %.10f\n", cur, myjg[cur], sum);
                sign = -sign;
            }
            printf("===============compute finish!=========== result: %.10f\n", sum);
        } else {
            pthread_mutex_unlock(&eventlock);
        }
    }
    return NULL;
}

主函式：線程建立與同步

int main() {
    pthread_mutex_init(&eventlock, NULL);
    pthread_cond_init(&myevent, NULL);

    printf("please input an integer:(<=%d) ", MAXS);
    while (scanf("%d", &max), max > MAXS) {
        printf("please input an integer:(<=%d) ", MAXS);
    }
    myjg[0] = 0;

    int i = 0;
    pthread_create(&threads[i++], NULL, myprint1, NULL); sleep(1);
    pthread_create(&threads[i++], NULL, myprint2, NULL); sleep(1);
    for (; i <= MAXTDS; i++) {
        pthread_create(&threads[i], NULL, mycomp, NULL);
        sleep(1);
    }

    sleep(MAXTDS * 2 * (i/10 + 1)); // 等待完成
    pthread_mutex_destroy(&eventlock);
    pthread_cond_destroy(&myevent);
    return 0;
}

執行輸出節選

please input an integer:(<=1000)26
time out
time out
1/1 finished,push 1.0000000000
1/2 finished,push 0.5000000000
...
1/1:1.0000000000 added result 1.0000000000
1/2:0.5000000000 added result 1.5000000000
...
*******1/1:1.0000000000 computed result 1.0000000000
*******1/2:0.5000000000 computed result 0.5000000000
...
===============compute finish!=========== result:0.7253718504

關鍵同步機制解析

常見坑點與最佳實踐

1. 條件變量必須配合互斥鎖：pthread_cond_wait/timedwait 入參鎖需已鎖定，函式內部原子性解鎖並休眠，被喚醒後重新加鎖返回。
2. 虛假喚醒：POSIX 允許無信號喚醒，必須用迴圈檢查條件（本例隱含於 ready 索引檢查）。
3. 廣播 vs 單播：多消費者皆需處理新資料用 broadcast；僅需一個處理用 signal，避免驚群效應。
4. 資源清理：主線程結束前 destroy 鎖與條件變量，避免洩漏。

編譯與執行

gcc -D_REENTRANT -lpthread -o threadpool threadpool.c
./threadpool

延伸方向

• 引入任務佇列，實現通用 ThreadPool::submit(task) 介面。
• 使用 pthread_condattr_setclock 指定單調時鐘，避免系統時間調整影響超時。
• 對比 epoll + 線程池（Reactor 模式）在高併發網路服務中的表現。