Linux C 生產者-消費者模型完整實戰：信號量、共享記憶體、多進程同步與互斥

編輯：關於C語言

問題背景

多進程環境下，生產者產生資料放入有限緩衝區，消費者從緩衝區取出處理。核心需求：

互斥：同一時間僅一進程存取緩衝區索引與資料。
同步：緩衝區滿時生產者阻塞；空時消費者阻塞。
資料共享：緩衝區、讀寫索引需跨進程可見。

核心 IPC 原語

原語	頭檔	關鍵函數	用途
System V 共享記憶體	`<sys/shm.h>`	`shmget/shmat/shmdt/shmctl`	跨進程高速資料交換
System V 信號量	`<sys/sem.h>`	`semget/semctl/semop`	進程同步/互斥 (PV 操作)
POSIX 信號量	`<semaphore.h>`	`sem_open/sem_wait/sem_post/sem_close/sem_unlink`	命名/未命名，介面更簡潔
POSIX 共享記憶體	`<sys/mman.h>`	`shm_open/mmap/munmap/shm_unlink`	檔案系統命名空間，支援 `mmap`

本文採用 **System V** 風格 (教學經典)，生產環境建議改用 **POSIX** (命名語意清晰、不需手動管理 key_t)。

資料結構設計


#define BUF_SIZE 10          // 緩衝區槽位數
#define SHM_KEY  0x1234      // 共享記憶體 key
#define SEM_KEY  0x5678      // 信號量集 key

// 共享記憶體佈局
struct SharedData {
    int buffer[BUF_SIZE];    // 環狀緩衝區
    int in;                  // 生產者寫入索引
    int out;                 // 消費者讀取索引
};

// 信號量集索引
enum { SEM_MUTEX = 0, SEM_EMPTY = 1, SEM_FULL = 2 };
// SEM_MUTEX: 互斥鎖 (初值 1)
// SEM_EMPTY: 空槽位數 (初值 BUF_SIZE)
// SEM_FULL:  滿槽位數 (初值 0)

封裝工具函數


#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>

// 信號量 P/V 操作
int sem_p(int semid, int index) {
    struct sembuf op = { index, -1, SEM_UNDO };
    return semop(semid, &op, 1);
}
int sem_v(int semid, int index) {
    struct sembuf op = { index, +1, SEM_UNDO };
    return semop(semid, &op, 1);
}

// 取得/建立共享記憶體
int GetShm(key_t key, size_t size, int flags) {
    int shmid = shmget(key, size, flags | 0666);
    if (shmid == -1 && errno == EEXIST) {
        shmid = shmget(key, 0, 0666); // 僅開啟
    }
    return shmid;
}

// 取得/建立信號量集
int GetSem(key_t key, int nsems, int flags) {
    int semid = semget(key, nsems, flags | 0666);
    if (semid == -1 && errno == EEXIST) {
        semid = semget(key, 0, 0666);
    }
    return semid;
}

// 初始化信號量值 (僅建立者呼叫)
void InitSem(int semid) {
    union semun { int val; struct semid_ds *buf; unsigned short *array; } arg;
    arg.val = 1; semctl(semid, SEM_MUTEX, SETVAL, arg);   // 互斥
    arg.val = BUF_SIZE; semctl(semid, SEM_EMPTY, SETVAL, arg); // 空槽
    arg.val = 0; semctl(semid, SEM_FULL, SETVAL, arg);    // 滿槽
}

生產者進程


#include "common.h"
#include <unistd.h>
#include <time.h>

int main() {
    // 1. 取得共享記憶體
    int shmid = GetShm(SHM_KEY, sizeof(struct SharedData), IPC_CREAT | IPC_EXCL);
    bool first = (shmid != -1);
    if (!first) shmid = GetShm(SHM_KEY, sizeof(struct SharedData), 0);
    struct SharedData *shm = shmat(shmid, NULL, 0);

    // 2. 取得信號量
    int semid = GetSem(SEM_KEY, 3, IPC_CREAT | IPC_EXCL);
    if (semid != -1) InitSem(semid); // 僅首個進程初始化

    // 3. 生產迴圈
    srand(time(NULL) ^ getpid());
    for (int i = 0; i < 20; i++) {
        int item = rand() % 1000;

        sem_p(semid, SEM_EMPTY);  // 等待空槽
        sem_p(semid, SEM_MUTEX);  // 進入臨界區

        shm->buffer[shm->in] = item;
        printf("[Producer %d] Produced %d at slot %d\n", getpid(), item, shm->in);
        shm->in = (shm->in + 1) % BUF_SIZE;

        sem_v(semid, SEM_MUTEX);  // 離開臨界區
        sem_v(semid, SEM_FULL);   // 通知有資料

        sleep(rand() % 3);
    }

    shmdt(shm);
    return 0;
}

消費者進程


#include "common.h"
#include <unistd.h>

int main() {
    int shmid = GetShm(SHM_KEY, sizeof(struct SharedData), 0);
    struct SharedData *shm = shmat(shmid, NULL, 0);
    int semid = GetSem(SEM_KEY, 3, 0);

    for (int i = 0; i < 20; i++) {
        sem_p(semid, SEM_FULL);   // 等待資料
        sem_p(semid, SEM_MUTEX);  // 進入臨界區

        int item = shm->buffer[shm->out];
        printf("[Consumer %d] Consumed %d from slot %d\n", getpid(), item, shm->out);
        shm->out = (shm->out + 1) % BUF_SIZE;

        sem_v(semid, SEM_MUTEX);
        sem_v(semid, SEM_EMPTY);  // 釋放空槽

        sleep(rand() % 2);
    }

    shmdt(shm);
    return 0;
}

編譯與執行


gcc -o producer producer.c -Wall
gcc -o consumer consumer.c -Wall

# 終端 1: 啟動 1 生產者
./producer
# 終端 2: 啟動 1 消費者
./consumer
# 觀察輸出交錯，生產者滿時阻塞，消費者空時阻塞

# 終端 3: 再啟動多個生產者/消費者驗證多進程同步
./producer & ./producer & ./consumer & ./consumer

關鍵細節深度解析

1. 信號量初始化競爭

多進程同時 IPC_CREAT | IPC_EXCL 建立信號量，僅一個成功 (回傳 id)，其餘 EEXIST 改為 IPC_CREAT 僅開啟。**僅建立成功者呼叫 InitSem**，避免重複初始化覆蓋計數。

2. `SEM_UNDO` 旗標

struct sembuf op = { ..., SEM_UNDO }; 讓核心記錄進程對信號量的調整值，進程異常終止 (未呼叫 sem_v) 時，核心自動回滾，防止死鎖。**生產環境強烈建議啟用**。

3. 環狀緩衝區索引管理


shm->in = (shm->in + 1) % BUF_SIZE;   // 生產者推進
shm->out = (shm->out + 1) % BUF_SIZE; // 消費者推進
// 滿判斷: (in + 1) % BUF_SIZE == out  (浪費 1 槽位區分空/滿)
// 或額外計數器

4. 共享記憶體權限與清理


// 建立時 0666 允許同用戶/組讀寫
// 程式正常結束前應清理 (示範簡略省略)
shmctl(shmid, IPC_RMID, NULL);  // 標記刪除，最後一個 detach 後實際釋放
semctl(semid, 0, IPC_RMID);     // 刪除信號量集
// 實際部署可用 shell 腳本或 systemd 服務單元管理生命週期

現代化替代方案：POSIX 命名信號量 + mmap


// POSIX 風格 (簡潔、命名、支援進程/執行緒雙模)
sem_t *mutex = sem_open("/my_mutex", O_CREAT, 0666, 1);
sem_t *empty = sem_open("/my_empty", O_CREAT, 0666, BUF_SIZE);
sem_t *full  = sem_open("/my_full",  O_CREAT, 0666, 0);

int fd = shm_open("/my_buffer", O_CREAT | O_RDWR, 0666);
ftruncate(fd, sizeof(struct SharedData));
struct SharedData *shm = mmap(NULL, sizeof(*shm), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);

// 使用 sem_wait/sem_post 替代 sem_p/sem_v
// 結束時 sem_close + sem_unlink + shm_unlink

優點：名字直觀、無 key_t 衝突、支援執行緒同步、自動清理語意清晰。

常見錯誤排查

現象	原因	解決
生產者/消費者均卡住不動	信號量初始值錯誤 (如 MUTEX=0, EMPTY=0)	檢查 `InitSem` 邏輯，確保 MUTEX=1, EMPTY=N, FULL=0
資料損壞/索引錯誤	缺少 MUTEX 保護，或 P/V 順序錯誤	嚴格遵循: P(EMPTY) → P(MUTEX) → 關鍵區 → V(MUTEX) → V(FULL)
進程死後信號量不釋放	未使用 SEM_UNDO，且無信號處理清理	加 SEM_UNDO；註冊 SIGTERM/SIGINT handler 呼叫 V 操作
多次執行殘留 IPC 物件	未清理 shm/sem	啟動腳本先 `ipcrm -a` 或使用 POSIX 命名 `_unlink`

結語

生產者-消費者是並發程式設計的「Hello World」，但正確處理互斥、同步、共享記憶體、錯誤恢復、資源清理卻蘊含系統程式設計的核心功力。掌握 System V 與 POSIX 兩套 IPC 機制的異同、信號量 PV 操作順序、SEM_UNDO 語意、環狀緩衝區索引管理、跨進程資源生命週期，即可在嵌入式、伺服器、分散式系統中自信構建高效、健壯的並發管道。