《Linux系統編程篇》System V信號量實現生產者與消費者問題（Linux 進程間通信（IPC））—

文章目錄

- - 📚 **生產者-消費者問題**
  - 🔑 **問題分析**
  - 🛠? **詳細實現：生產者-消費者**
  - - **步驟 1：定義信號量和緩沖區**
    - **步驟 2：創建信號量**
    - **步驟 3：生產者進程**
    - **步驟 4：消費者進程**
    - **步驟 5：創建進程并啟動**
  - 🧑?🔧 **完整代碼示例**
  - 🎯 **關鍵點總結**

接上節，我們來詳細展開一下 生產者-消費者問題，并用 System V 信號量 來解決它。這個經典問題幫助我們理解如何在多個進程間同步和互斥地共享資源。

📚 生產者-消費者問題

生產者-消費者問題是多進程同步問題中的經典例子。問題的背景是：有兩個進程，一個生產者（Producer）不斷生產產品，另一個消費者（Consumer）不斷消費產品。兩者都需要共享一個有限的緩沖區。生產者往緩沖區寫入數據，消費者從緩沖區讀取數據。為了避免并發問題，我們需要同步生產者和消費者的訪問。

具體的挑戰是：

互斥：生產者和消費者在訪問共享緩沖區時，不能同時操作。
同步：緩沖區不能超過最大容量，也不能為空。

🔑 問題分析

我們需要使用信號量來解決這些問題，具體來說，我們需要：

一個信號量來控制緩沖區的空位置數（空位信號量）。
一個信號量來控制緩沖區的已滿位置數（已滿信號量）。
一個互斥信號量來保證每次只有一個進程（生產者或消費者）可以訪問緩沖區。

我們通過信號量來控制：

當緩沖區為空時，消費者應該等待。
當緩沖區已滿時，生產者應該等待。
互斥信號量保證在訪問共享緩沖區時，只有一個進程能夠進入臨界區。

🛠? 詳細實現：生產者-消費者

步驟 1：定義信號量和緩沖區

我們將使用以下信號量：

empty：緩沖區中空位的數量，初始值為 BUFFER_SIZE。
full：緩沖區中已滿的數量，初始值為 0。
mutex：互斥鎖，用來確保每次只有一個進程能夠訪問緩沖區，初始值為 1。

緩沖區本身可以用一個數組來表示：

#define BUFFER_SIZE 5  // 緩沖區大小
#define NUM_ITEMS 10   // 生產和消費的物品數量int buffer[BUFFER_SIZE];  // 緩沖區
int in = 0;  // 指向下一個要寫入的位置
int out = 0; // 指向下一個要讀取的位置

步驟 2：創建信號量

我們通過 semget() 創建信號量集：

int sem_id = semget(IPC_PRIVATE, 3, IPC_CREAT | 0666);  // 創建3個信號量
if (sem_id == -1) {perror("semget");exit(1);
}// 初始化信號量
semctl(sem_id, 0, SETVAL, BUFFER_SIZE);  // empty 信號量：初始為緩沖區大小
semctl(sem_id, 1, SETVAL, 0);  // full 信號量：初始為0，表示緩沖區沒有物品
semctl(sem_id, 2, SETVAL, 1);  // mutex 信號量：初始為1，表示可以訪問緩沖區

步驟 3：生產者進程

生產者進程的工作流程如下：

等待空位信號量（empty）：只有在有空位時才能生產。
獲取互斥信號量（mutex）：進入臨界區，確保沒有其他進程操作緩沖區。
生產：將數據放入緩沖區。
釋放互斥信號量（mutex）：退出臨界區。
增加已滿信號量（full）：表明緩沖區中有一個新產品，消費者可以消費。

生產者代碼示例：

void producer(int sem_id) {for (int i = 0; i < NUM_ITEMS; i++) {struct sembuf sops[2];// P(empty)sops[0].sem_num = 0;sops[0].sem_op = -1;sops[0].sem_flg = 0;// P(mutex)sops[1].sem_num = 2;sops[1].sem_op = -1;sops[1].sem_flg = 0;semop(sem_id, sops, 2);buffer[in] = i;printf("生產者生產了產品 %d\n", i);in = (in + 1) % BUFFER_SIZE;// V(mutex) 和 V(full)struct sembuf sops_release[2];sops_release[0].sem_num = 2; // mutexsops_release[0].sem_op = 1;sops_release[0].sem_flg = 0;sops_release[1].sem_num = 1; // fullsops_release[1].sem_op = 1;sops_release[1].sem_flg = 0;semop(sem_id, sops_release, 2);sleep(1);}
}

步驟 4：消費者進程

消費者進程的工作流程如下：

等待已滿信號量（full）：只有在緩沖區有物品時才能消費。
獲取互斥信號量（mutex）：進入臨界區，確保沒有其他進程操作緩沖區。
消費：從緩沖區中取出數據。
釋放互斥信號量（mutex）：退出臨界區。
增加空位信號量（empty）：表明緩沖區有一個空位，生產者可以生產。

消費者代碼示例：

void consumer(int sem_id) {for (int i = 0; i < NUM_ITEMS; i++) {struct sembuf sops[2];// P(full)sops[0].sem_num = 1;sops[0].sem_op = -1;sops[0].sem_flg = 0;// P(mutex)sops[1].sem_num = 2;sops[1].sem_op = -1;sops[1].sem_flg = 0;semop(sem_id, sops, 2);int item = buffer[out];printf("消費者消費了產品 %d\n", item);out = (out + 1) % BUFFER_SIZE;// V(mutex) 和 V(empty)struct sembuf sops_release[2];sops_release[0].sem_num = 2; // mutexsops_release[0].sem_op = 1;sops_release[0].sem_flg = 0;sops_release[1].sem_num = 0; // emptysops_release[1].sem_op = 1;sops_release[1].sem_flg = 0;semop(sem_id, sops_release, 2);sleep(1);}
}

步驟 5：創建進程并啟動

在 main() 函數中，我們創建了一個子進程，用于運行消費者進程。父進程將作為生產者運行。

int main() {int sem_id = semget(IPC_PRIVATE, 3, IPC_CREAT | 0666);if (sem_id == -1) {perror("semget");exit(1);}semctl(sem_id, 0, SETVAL, BUFFER_SIZE); // emptysemctl(sem_id, 1, SETVAL, 0);           // fullsemctl(sem_id, 2, SETVAL, 1);           // mutexpid_t pid = fork();if (pid == 0) {consumer(sem_id);} else if (pid > 0) {producer(sem_id);wait(NULL);semctl(sem_id, 0, IPC_RMID);} else {perror("fork failed");exit(1);}return 0;
}

在這段代碼中：

fork()：通過 fork() 創建一個新的子進程。父進程作為生產者執行 producer() 函數，子進程作為消費者執行 consumer() 函數。
父進程和子進程分工：生產者不斷生產物品放入緩沖區，消費者從緩沖區取出物品進行消費。
wait(NULL)：父進程使用 wait() 來等待子進程的結束，這樣可以確保父進程在子進程完成后再退出，避免資源的提前釋放。
刪除信號量集：為了避免信號量集泄露，程序結束時通過 semctl() 刪除創建的信號量集。

🧑?🔧 完整代碼示例

這里是完整的代碼，包含了生產者和消費者進程的實現，以及使用 System V 信號量同步和互斥訪問共享緩沖區。

#include <stdio.h>
#include <stdlib.h>
#include <sys/sem.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>#define BUFFER_SIZE 5
#define NUM_ITEMS 10int buffer[BUFFER_SIZE];
int in = 0;
int out = 0;void producer(int sem_id) {for (int i = 0; i < NUM_ITEMS; i++) {struct sembuf sops[2];// P(empty)sops[0].sem_num = 0;sops[0].sem_op = -1;sops[0].sem_flg = 0;// P(mutex)sops[1].sem_num = 2;sops[1].sem_op = -1;sops[1].sem_flg = 0;semop(sem_id, sops, 2);buffer[in] = i;printf("生產者生產了產品 %d\n", i);in = (in + 1) % BUFFER_SIZE;// V(mutex) 和 V(full)struct sembuf sops_release[2];sops_release[0].sem_num = 2; // mutexsops_release[0].sem_op = 1;sops_release[0].sem_flg = 0;sops_release[1].sem_num = 1; // fullsops_release[1].sem_op = 1;sops_release[1].sem_flg = 0;semop(sem_id, sops_release, 2);sleep(1);}
}void consumer(int sem_id) {for (int i = 0; i < NUM_ITEMS; i++) {struct sembuf sops[2];// P(full)sops[0].sem_num = 1;sops[0].sem_op = -1;sops[0].sem_flg = 0;// P(mutex)sops[1].sem_num = 2;sops[1].sem_op = -1;sops[1].sem_flg = 0;semop(sem_id, sops, 2);int item = buffer[out];printf("消費者消費了產品 %d\n", item);out = (out + 1) % BUFFER_SIZE;// V(mutex) 和 V(empty)struct sembuf sops_release[2];sops_release[0].sem_num = 2; // mutexsops_release[0].sem_op = 1;sops_release[0].sem_flg = 0;sops_release[1].sem_num = 0; // emptysops_release[1].sem_op = 1;sops_release[1].sem_flg = 0;semop(sem_id, sops_release, 2);sleep(1);}
}int main() {int sem_id = semget(IPC_PRIVATE, 3, IPC_CREAT | 0666);if (sem_id == -1) {perror("semget");exit(1);}semctl(sem_id, 0, SETVAL, BUFFER_SIZE); // emptysemctl(sem_id, 1, SETVAL, 0);           // fullsemctl(sem_id, 2, SETVAL, 1);           // mutexpid_t pid = fork();if (pid == 0) {consumer(sem_id);} else if (pid > 0) {producer(sem_id);wait(NULL);semctl(sem_id, 0, IPC_RMID);} else {perror("fork failed");exit(1);}return 0;
}