PHP 消息隊列實現 Android 與 Web 即時通訊

PHP 消息隊列實現 Android 與 Web 即時通訊

場景：Android 客戶端通過 Socket 推送數據到服務端，Web 前端通過 Server-Sent Events (SSE) 即時接收展示。核心難點在於 PHP 進程間通訊 (IPC)：Socket 接收進程與 SSE 推送進程如何共享數據？本文採用 System V 消息隊列作為橋梁。

架構圖解

Android (Socket Client)
        ↓ TCP
PHP Socket Server (常駐進程) → 消息隊列 (msg_queue)
        ↑                           ↓
PHP SSE Endpoint (Web 請求) ← 讀取隊列推送給瀏覽器

1. Android 端 Socket 發送（子線程）

// 核心發送邏輯
private void sendMsg(String data) {
    if (!connected) return;
    synchronized (socket) {
        out.println(data); // PrintWriter 自動 flush
    }
}

2. PHP Socket 服務端（接收寫入隊列）

// socket_server.php
function buildSocket($msgQueue) {
    $addr = '0.0.0.0'; $port = 54321;
    $sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
    socket_bind($sock, $addr, $port);
    socket_listen($sock, 4);
    echo "Listening on $port...\n";

    $client = socket_accept($sock);
    while (($buf = socket_read($client, 8192)) !== false) {
        if (strlen($buf) > 1) {
            msg_send($msgQueue, 1, $buf); // 寫入隊列，類型 1
        }
    }
    socket_close($client);
    socket_close($sock);
}

$key = ftok(__FILE__, 's');
$queue = msg_get_queue($key);
buildSocket($queue);

3. SSE 端點（讀取隊列推送給前端）

// sse.php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

$key = ftok('socket_server.php', 's'); // 必須與寫入端 key 一致
$queue = msg_get_queue($key, 0666);

echo "data: connected\n\n";
flush();

while (true) {
    $stat = msg_stat_queue($queue);
    if ($stat['msg_qnum'] == 0) {
        usleep(100000); // 100ms 輪詢
        continue;
    }
    msg_receive($queue, 0, $type, 1024, $msg, true, MSG_IPC_NOWAIT);
    echo "data: $msg\n\n";
    flush();
}

4. 前端接收（EventSource）

const evt = new EventSource('sse.php');
evt.onmessage = e => {
    document.getElementById('msg').textContent = e.data;
};

關鍵點與避坑

• ftok 路徑一致性：讀寫端必須用同一文件路徑與 project id 生成 key。
• 隊列阻塞：消息積壓達上限會阻塞寫入，生產環境建議用 Redis/RabbitMQ 替代。
• SSE 限制：單域名連接數限制（HTTP/1.1 為 6），IE 不支援，生產可考慮 WebSocket + Swoole/Workerman。
• 進程管理：Socket 服務端需用 Supervisor/Systemd 守護常駐。