Linux IPX 訊息佇列實作程式間資料交換

– 10 分鐘閱讀

在 Linux 系統中，程式間通訊（IPC）是實作多個程式協同工作的重要機制。IPX 訊息佇列作為一種 IPC 方式，允許不同程式以非同步方式交換資料。透過 msgget 函式建立訊息佇列，並使用 msgsnd 和 msgrcv 函式分別進行訊息的傳送和接收。訊息結構體包含訊息型別 mtype 和訊息內容 mtext，mtype 可用於區分不同型別的訊息，方便接收端進行過濾處理。msgsnd 函式的 msgflg 引數可設定非阻塞模式，避免傳送端因佇列滿而阻塞。接收端使用 msgrcv 函式接收訊息，msgtyp 引數可以設定接收特定型別的訊息，提升效率。使用完畢後，應使用 msgctl 函式銷毀訊息佇列，釋放系統資源。除了訊息佇列，分享記憶體也是一種常用的 IPC 方式，允許多個程式分享同一塊記憶體區域，實作高效的資料交換。然而，分享記憶體需要考慮程式同步問題，避免資料競爭。訊號量作為一種同步機制，可以有效控制程式對分享資源的存取，防止死鎖和競爭條件的發生。shmget 函式用於建立分享記憶體區段，shmat 和 shmdt 函式分別用於裝載和解除安裝分享記憶體區段到程式的位址空間。shmctl 函式則用於控制分享記憶體區段的屬性，例如許可權和鎖定狀態。

使用Linux IPX 訊息佇列進行資料交換

訊息佇列的基本操作

在Linux系統中，訊息佇列（Message Queue）是一種程式間通訊（IPC）機制，允許不同程式之間進行資料交換。訊息佇列的操作主要包括建立、傳送、接收和銷毀訊息佇列。以下將詳細介紹這些操作的具體實作方法。

建立訊息佇列

訊息佇列的建立使用msgget函式，該函式會傳回一個訊息佇列識別碼（ID）。以下是msgget函式的原型：

int msgget(key_t key, int msgflg);
  • key：用於識別訊息佇列的鍵值。
  • msgflg：控制許可權和行為的標誌，例如IPC_CREAT用於建立新的訊息佇列。

在成功建立訊息佇列後，相關引數會被初始化，例如：

  • msg_qnum：設定為0，表示當前沒有訊息。
  • msg_lspid 和 msg_lrpid：設定為0，表示沒有最後傳送或接收的程式ID。
  • msg_stime 和 msg_rtime：設定為0，表示沒有傳送或接收的時間戳。
  • msg_perm.uid 和 msg_perm.cuid：設定為程式的有效使用者ID。
  • msg_perm.gid 和 msg_perm.cgid：設定為程式的有效群組ID。

傳送訊息到佇列

傳送訊息到訊息佇列使用msgsnd函式。以下是其原型：

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
  • msqid：由msgget函式傳回的訊息佇列識別碼。
  • msgp：指向要傳送的訊息結構體的指標。
  • msgsz：訊息的大小（以位元組為單位）。
  • msgflg：控制行為的標誌，例如IPC_NOWAIT用於非阻塞模式。

訊息結構體定義如下：

struct msgbuf {
    long mtype;
    char mtext[1];
};
內容解密：

此段落程式碼定義了一個名為 msgbuf 的結構體，其中包含兩個成員：

  1. mtype: 這是一個長整數型變數，用於標識訊息的型別。這個型別可以用來區分不同型別的訊息，便於接收端根據型別進行過濾或處理。
  2. mtext: 這是一個字元陣列，用於儲存實際的訊息內容。注意陣列大小設定為1，實際使用時需要根據需要調整大小。

如果系統中已經存在太多訊息、超過了預設限制或設定了非阻塞模式且無法立即傳送，則會傳回錯誤程式碼。

範例程式碼

以下是一個簡單的範例程式碼，展示如何建立一個訊息佇列並向其中傳送訊息：

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

struct msgbuf {
    long mtype;
    char mtext[200];
};

int main(void) {
    struct msgbuf buf;
    int msqid;
    key_t key = 100;

    if ((msqid = msgget(key, 0644 | IPC_CREAT)) == -1) {
        perror("msgget");
        exit(1);
    }

    printf("Get message from keyboard\n");
    buf.mtype = 1; /* 訊息型別設定為1 */
    while (fgets(buf.mtext, sizeof(buf.mtext), stdin) != NULL) {
        if (msgsnd(msqid, (struct msgbuf *)&buf, sizeof(buf), 0) == -1) {
            perror("message not sent properly");
        }
    }

    if (msgctl(msqid, IPC_RMID, NULL) == -1) {
        perror("message queue destroyed");
        exit(1);
    }

    return 0;
}
內容解密：

此段落程式碼展示如何建立一個訊息佇列並向其中傳送來自鍵盤輸入的訊息。以下是詳細說明：

  1. 結構體定義：

    • 定義了一個名為 msgbuf 的結構體，包含兩個成員：長整數型變數 mtype 和字元陣列 mtext。

  2. 主程式流程：

    • 初始化鍵值並使用 msgget() 函式建立或取得一個訊息佇列。若失敗則輸出錯誤訊息並離開程式。
    • 提示使用者輸入訊息。
    • 使用 fgets() 函式從標準輸入讀取訊息並存入結構體中的 mtext 成員中。
    • 使用 msgsnd() 函式將訊息傳送到訊息佇列中。若失敗則輸出錯誤訊息。
    • 在結束時使用 msgctl() 函式刪除訊息佇列。

接收訊息

接收訊息使用sgrcv()函式。以下是其原型：

int msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
  • msqid: 訊息佇列識別碼。
  • msgp: 指向用於儲存接收到的訊息之緩衝區指標。
  • msgsz: 需要接收到訊號之緩衝區長度（以位元組計算）。
  • msgtyp: 用來決定要接收哪一種型別之訊號。它可以是正整數、零或負整數：
    • 零: 接收第一筆任意型別之訊號。
    • 正整數: 接收第一筆具有指定型別之訊號。
    • 負整數: 接收第一筆具有小於或等於其絕對值之訊號型別。

範例程式碼

以下範例展示瞭如何從消費物件中讀取訊號並將其顯示在螢幕上：

#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

struct msgbuf {
    long mtype;
    char mtext[200];
};

int main(void) {
    struct msgbuf buf;
    int msqid;
    key_t key = 41;

    if ((msqid = msgget(key, 0644)) == -1) { /* connect to the queue */
        perror("message queue does not exist ");
        exit(1);
    }

    printf("Receiving messages from message queue \n");

    for (;;) {
        if (msgrcv(msqid, (struct msgbuf *)&buf, sizeof(buf), 0, 0) == -1) {
            perror("msgrcv");
            exit(1);
        }
        printf("%s\n", buf.mtext);
    }

    return 0;
}
內容解密：

此段落程式碼展示瞭如何從一個已存在的消費物件中接收訊號並顯示在螢幕上。以下是詳細說明：

  1. 結構體定義：

    • 再次定義了一個名為 msgbuf 的結構體，包含兩個成員：長整數型變數 mtype 和字元陣列 mtext.

  2. 主程式流程：

    • 初始化鍵值並使用 msgget() 函式連線到已存在的消費物件。若失敗則輸出錯誤訊號並離開程式。
    • 提示使用者正在從訊號佇列中接收訊號。
    • 構成無限迴圈不斷執行接收訊號動作：
      • 使用 msgrcv() 函式從訊號佇列中接收訊號並將其儲存在結構體中的 mtext 中。若失敗則輸出錯誤訊號並離開程式。
      • 陳述式將從佇列中所讀取到之訊號內容顯示在螢幕上。

銷毀和控制消費物件

銷毀

銷毀一個已經存在之訊號佇列可以使用到下面這個陳述式：

int msgctl(int cmd, struct msqid_ds *buf);
主要引數:
  • cmd: 用來指定命令值（例如 IPC_RMID 用來移除訊號佇列）。
  • buf: 指向一個指令結果物件結構。

只有擁有有效特權身份（例如超級管理員）或具有合法許可權之進行者才能成功執行此命令。

控制

透過控制命令可以改變訊號佇列中的各種屬性，例如許可權、大小等。這裡我們給出原型如下:

int msgctl(int msqid, int cmd, struct msqid_ds *buf);
主要引數:
  • msqid: 消費物件識別碼（由 msgget() 生成）。
  • cmd: 指令引數值（例如 IPC_STAT 用來取得狀態資訊）。
  • buf: 指向一個指令結果物件結構。

這些控制命令使得開發人員能夠靈活地管理和操作訊號佇列。

其他考量

改善方式與技術選擇

從開發角度來說，除了傳統 IPC 機制外還可考慮採用現代技術如Redis、RabbitMQ等替代方案進行資料傳遞。它們具有更高效能、更豐富功能以及更易維護等優點。

未來趨勢預測

隨著微服務架構在企業中的普及，分散式系統架構設計成為趨勢之一；同時資料交換與管理也將隨著需要逐步走向更加複雜與精細化。而 IPC 機制在高效能運算場景下依然具有重要意義與應用價值。

個人獨特見解

玄貓認為雖然目前 Linux IPC 技術已經相對成熟且穩定；然而在面臨現代化需求下可能會遇到一些瓶頸與挑戰。因此玄貓認為未來應更加註重新興技術與舊有技術之間如何進行深度融合與最佳化配合；同時也需不斷提升其安全性與可靠性以滿足日益增加之複雜業務需求。

分享記憶體與程式同步化

分享記憶體是一種用於程式間通訊（IPC）的技術，允許多個程式透過讀寫分享的記憶體區域來交換資訊。在這種方式下，各個程式需要協調對分享記憶體區段的使用，以避免資料衝突或競爭條件。

分享記憶體的建立與管理

建立分享記憶體

在 UNIX/Linux 作業系統中，使用 shmget 函式來取得對分享記憶體區段的存取許可權。shmget 函式的原型如下：

int shmget(key_t key, size_t size, int shmflg);
  • key：是一個與分享記憶體相關聯的存取值。
  • size：是請求的分享記憶體區段的大小（以位元組為單位）。
  • shmflg：指定初始存取許可權和建立控制旗標。

當 shmget 呼叫成功時，它會傳回分享記憶體區段的 ID。這個 ID 可以用來操作該分享記憶體區段。

控制分享記憶體區段

使用 shmctl 函式來改變分享記憶體區段的許可權和其他特性。shmctl 函式的原型如下：

int shmctl(int shmid, int cmd, struct shmid_ds *buf);
  • shmid：是要操作的分享記憶體區段的 ID。
  • cmd：是控制命令，可以是以下幾種：
    • SHM_LOCK：將指定的分享記憶體區段鎖定在記憶體中。
    • SHM_UNLOCK：解鎖分享記憶體區段。
    • IPC_STAT：傳回控制結構中的狀態資訊並放置在由 buf 指向的緩衝區中。
    • IPC_SET：設定有效的使用者和群組識別碼以及存取許可權。
    • IPC_RMID：移除分享記憶體區段。
  • buf：是一個指向結構型別 struct shmid_ds 的指標，定義在 <sys/shm.h> 中。

裝載與解除安裝分享記憶體

使用 shmat 和 shmdt 函式來裝載和解除安裝分享記憶體區段到呼叫程式的位址空間。其原型如下：

void *shmat(int shmid, const void *shmaddr, int shmflg);
int shmdt(const void *shmaddr);
  • shmat 函式將與分享記憶體識別碼 shmid 相關聯的分享記憶體區段裝載到呼叫程式的位址空間中，並傳回一個指向該區段起始位置的指標。
  • shmdt 函式則將位於由 shmaddr 指定位址處的分享記憶體區段解除安裝。

應用範例

以下是一個簡單的應用範例，展示如何使用分享記憶體進行程式間通訊。這個範例包括一個伺服器程式和一個客戶端程式。

伺服器程式

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <stdio.h>
#include <string.h>

#define SHMSIZE 100
#define SHM_KEY 1267

int main() {
    char *str;
    int shmid;
    key_t key;
    char *shm, *s;
    key = SHM_KEY;

    /* 建立分享記憶體區段 */
    if ((shmid = shmget(key, SHMSIZE, IPC_CREAT | 0666)) < 0) {
        printf("Segment not created successfully\n");
        return 1;
    }

    /* 裝載分享記憶體區段 */
    if ((shm = shmat(shmid, NULL, 0)) == (char *) -1) {
        perror("shmat");
        return 1;
    }

    s = shm;
    printf("Enter a string: ");
    scanf("%s", str);
    strcpy(s, str);
    *s = '\0';

    /* 鎖住並等待客戶端讀取 */
    while (*shm != '*')
        sleep(1);

    return 0;
}

客戶端程式

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <stdio.h>

#define SHMSIZE 100
#define SHM_KEY 1267

int main() {
    int shmid;
    key_t key;
    char *shm, *s;

    if ((shmid = shmget(SHM_KEY, SHMSIZE, 0666)) < 0) {
        printf("\n Memory segment not created successfully \n");
        return 1;
    }

    if ((shm = shmat(shmid, NULL, 0)) == (char *) -1) {
        perror("shmat");
        return 1;
    }

    /* 読取伺服器寫入的資料 */
    for (s = shm; *s != '\0'; s++)
        putchar(*s);
    putchar('\n');

    /* 在第一個位置放置 '*' */
    *shm = '*';

    return 0;
}

資源同步化

在多個程式同時存取同一資源時，需要進行同步化以避免系統死鎖。UNIX 作業系統利用訊號量（semaphore）來控制對於分享資源的存取。

說明什麼是訊號量

訊號量本身並不是 IPC 的形式，而是一種控制多個程式對於分享資源存取的變數。訊號量包含了一個整數值，允許程式透過檢查和設定這個值來進行同步化操作。這樣可以確保檢查和設定操作不會被其他程式幹擾。

操作步驟：
  1. 檢查訊號量：程式檢查控制存取資源的訊號量。
  2. 判斷值：如果訊號量值為正，程式可以存取資源並將值減少1；如果為零，則程式進入睡眠狀態直到訊號量還原為正。
  3. 完成功能後釋放：完成對資源的使用後，程式將訊號量值加回去，讓其他等待中的程式能夠獲得資源。

命令解說：

wait 操作

wait 操作會先檢查訊號量值是否為預期值；如果是，則減少其值並傳回；否則將阻塞呼叫者直到訊號量達到所需值。

signal 操作

signal 操作會增加訊號量值以表示資源已釋放，使其他等待中的程式能夠繼續執行。

此圖示展示了兩個程式如何透過使用訊號量來協調對於分享資源的存取：

  graph TD
    F[F]
    F2[F2]
    H[H]
    H2[H2]
A[Process A] --> B{Check Semaphore}
B -->|Value > 0| C[Use Resource]
B -->|Value = 0| D[Sleep]
C --> E[Decrement Semaphore]
E --> F{Resource Used?}
F -->|Yes| G[Increment Semaphore]
F -->|No| C
D --> H{Semaphore Value Changed?}
H -->|Yes| B
H -->|No| D
B -- "Process B" --> B2{Check Semaphore}
B2 -->|Value > 0| C2[Use Resource]
B2 -->|Value = 0| D2[Sleep]
C2 --> E2[Decrement Semaphore]
E2 --> F2{Resource Used?}
F2 -->|Yes| G2[Increment Semaphore]
F2 -->|No| C2
D2 --> H2{Semaphore Value Changed?}
H2 -->|Yes| B2
H2 -->|No| D2

內容解密：

此圖示展示了兩個進行（Process A 和 Process B）如何透過訊號量來協調對於分享資源（例如印表機、檔案、資料函式庫等）的存取。兩個進行都會首先檢查（Check Semaphore）訊號量當前值：

  • 若值大於零表示有可用單位資源，進行便可直接使用（Use Resource），同時減少（Decrement Semaphore）訊號量單位。
  • 若值等於零表示沒有可用單位資源，進行便會進入睡眠狀態（Sleep），直到訊號量變化（Semaphore Value Changed?）。
  • 一旦完成使用（Resource Used?），進行將增加（Increment Semaphore）訊號量單位以通知其他等待中的進行該可再次使用該資源。這樣透過訊號量這種機制能有效避免競爭條件及死鎖問題。

在實務應用中，以上技術與方法能有效協助多重進行在作業系統中平行執行並且高效地進行彼此間協調工作任務。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。