PHP System V 訊息佇列實戰：進程間通訊完整測試範例

簡介

System V 訊息佇列（msg_* 函數族）提供核心級的進程間通訊機制，適合：

• CLI 長期運行進程間命令傳遞
• Web 請求與背景 Worker 解耦
• 多進程任務分發與結果收集

完整雙進程測試腳本

單一檔案支援 ?type=send 發送端與 ?type=receive 接收端：

0）
$maxSize = 1024;    // 最大訊息長度
$serialize = false; // 是否序列化
$blockSend = false; // 佇列滿時是否阻塞

// 發送模式
if (isset($_GET['type']) && $_GET['type'] === 'send') {
    echo ">>> 進入發送模式 (Ctrl+C 停止)\n";
    for ($i = 0; $i < 20; $i++) {
        $msg = "Hello from Sender #$i at " . date('H:i:s');
        $err = null;
        if (msg_send($queue, $sendType, $msg, $serialize, $blockSend, $err)) {
            echo "✓ 發送 #$i: $msg\n";
        } else {
            echo "✗ 發送失敗 #$i: ";
            var_dump($err);
        }
        sleep(1);
    }
    echo ">>> 發送完成\n";
}
// 接收模式
else {
    echo ">>> 進入接收模式 (每秒輪詢)\n";
    for ($i = 0; $i < 20; $i++) {
        sleep(1);
        $stat = msg_stat_queue($queue);
        echo "佇列狀態: 訊息數={$stat['msg_qnum']}, 佇列位元組={$stat['msg_qbytes']}\n";

        if ($stat['msg_qnum'] > 0) {
            $msgType = 0; // 0 表示接收任意類型最早訊息
            $receivedType = 0;
            $data = '';
            $err = null;
            if (msg_receive($queue, $msgType, $receivedType, $maxSize, $data, $serialize, MSG_IPC_NOWAIT, $err)) {
                echo "✓ 接收 [類型:$receivedType] $data\n";
            } else {
                echo "✗ 接收失敗: ";
                var_dump($err);
            }
        } else {
            echo "  (佇列空)\n";
        }
    }
    echo ">>> 接收測試結束\n";
}

執行步驟與預期結果

終端機 1：啟動接收端

php msg_test.php?type=receive

終端機 2：啟動發送端

php msg_test.php?type=send

接收端輸出範例

=== System V 訊息佇列測試 ===
>>> 進入接收模式 (每秒輪詢)
佇列狀態: 訊息數=0, 佇列位元組=16384
  (佇列空)
佇列狀態: 訊息數=1, 佇列位元組=16384
✓ 接收 [類型:1] Hello from Sender #0 at 14:30:01
佇列狀態: 訊息數=1, 佇列位元組=16384
✓ 接收 [類型:1] Hello from Sender #1 at 14:30:02
...

進階用法：多類型訊息路由

// 發送不同類型訊息
msg_send($queue, 1, 'UserLogin:uid123');    // 類型 1：用戶事件
msg_send($queue, 2, 'OrderCreate:ord456');  // 類型 2：訂單事件
msg_send($queue, 3, 'PaymentDone:pay789');  // 類型 3：支付事件

// 接收端依類型分派
while (true) {
    // 接收類型 1 或 2（優先處理用戶/訂單）
    if (msg_receive($queue, 1, $type, 1024, $data, false, MSG_IPC_NOWAIT)) {
        dispatch($type, $data);
        continue;
    }
    if (msg_receive($queue, 2, $type, 1024, $data, false, MSG_IPC_NOWAIT)) {
        dispatch($type, $data);
        continue;
    }
    // 再處理類型 3
    if (msg_receive($queue, 3, $type, 1024, $data, false, MSG_IPC_NOWAIT)) {
        dispatch($type, $data);
    }
    usleep(100000); // 100ms 避免佔滿 CPU
}

關鍵參數說明

常見問題排查

Q: msg_get_queue 失敗 Permission denied

A: 檢查 /proc/sys/kernel/msgmni、msgmax、msgmnb 核心參數；或嘗試 ipcs -q 看現有佇列權限

Q: 訊息內容被截斷

A: maxsize 參數必須 ≥ 實際訊息長度；系統限制 msgmax（通常 8KB），大訊息請改用共享記憶體+訊號通知

Q: 如何跨機器通訊？

A: System V IPC 僅限單機。跨機請用 Redis Streams、RabbitMQ、Kafka、gRPC 等網路協議

生產環境建議封裝類

class MsgQueue {
    private $queue;
    private $defaultType = 1;

    public function __construct(string $name, int $perms = 0666) {
        $key = ftok(__DIR__ . '/.ipc_key', $name[0]);
        $this->queue = msg_get_queue($key, $perms);
    }

    public function push(string $data, int $type = 0): bool {
        $type = $type ?: $this->defaultType;
        return msg_send($this->queue, $type, json_encode($data), false, false);
    }

    public function pop(int $type = 0, bool $wait = false): ?array {
        $flags = $wait ? 0 : MSG_IPC_NOWAIT;
        $msgType = 0;
        $data = '';
        $ok = msg_receive($this->queue, $type, $msgType, 8192, $data, false, $flags);
        return $ok ? ['type' => $msgType, 'data' => json_decode($data, true)] : null;
    }

    public function stats(): array {
        return msg_stat_queue($this->queue);
    }
}