ThinkPHP5 + Workerman如何实现广播消息到所有用户？

Question

在使用 ThinkPHP5 集成 Workerman 实现 WebSocket 服务时，开发者常遇到“如何将消息广播给所有在线用户”的问题。典型场景如实时通知、聊天室等。常见问题是：每次通过 `foreach` 遍历 `$connection->worker->connections` 发送数据时，仅部分客户端收到消息，或连接未正确识别。其根源在于未正确管理连接实例，或未对有效连接进行鉴权与存储。此外，Workerman 的 `onMessage` 回调中若未保存用户连接标识，会导致广播时无法精准推送。如何在 ThinkPHP5 中结合 Workerman 实现稳定、高效的全量用户广播，成为开发中的关键技术难点。

Answer 1

ThinkPHP5 集成 Workerman 实现 WebSocket 全量广播：从基础到高可用架构设计

1. 问题背景与典型场景分析

在现代 Web 应用中，实时通信已成为刚需。使用 ThinkPHP5 集成 Workerman 构建 WebSocket 服务，是中小型项目实现长连接通信的常见方案。典型应用场景包括：

• 实时通知系统（如订单状态更新）
• 在线聊天室或客服系统
• 多人协作白板或文档编辑
• 股票行情推送、赛事直播弹幕等高并发场景

然而，在实际开发中，开发者频繁遇到“消息无法广播给所有用户”的问题。尽管代码逻辑看似正确——通过遍历 $connection->worker->connections 发送数据，但结果往往是部分客户端收不到消息，甚至出现连接丢失或重复发送。

2. 常见问题排查路径

问题现象	可能原因	调试建议
仅部分用户收到消息	连接未鉴权，无效连接参与广播	检查 onMessage 中是否建立 user_id 与 connection 的映射
新用户上线后无法接收历史通知	广播逻辑未覆盖刚建立连接的用户	确认 onConnect/onMessage 是否完成身份绑定
断线重连后消息中断	连接标识未持久化或未重新注册	引入 Redis 存储连接状态
内存泄漏、Worker 进程崩溃	connection 对象被长期引用未释放	监控 Worker 内存使用，设置最大生命周期

3. 核心机制解析：Workerman 的连接管理模型

Workerman 启动多个 Worker 进程，每个进程独立维护自己的 $connections 属性，类型为 ConnectionInterface[]。关键点在于：

// 获取当前 Worker 所有活动连接
foreach ($worker->connections as $conn) {
    $conn->send('Hello');
}
    

上述代码只能广播本进程内的连接。若部署多进程（如 4 个 Worker），则每个进程只能触达其管辖的连接子集，导致“部分用户收到消息”。

4. 单机环境下全量广播实现方案

解决方案是在 onMessage 回调中完成用户身份绑定，并维护一个全局连接池。示例如下：

use Workerman\Worker;
use Workerman\WebServer;
use GatewayWorker\Gateway;

$worker = new Worker('websocket://0.0.0.0:8080');

// 存储用户连接 map: user_id => connection
$worker->uidConnections = [];

$worker->onConnect = function($connection) use ($worker) {
    echo "New connection from {$connection->remoteAddress}\n";
};

$worker->onMessage = function($connection, $data) use ($worker) {
    $msg = json_decode($data, true);

    // 客户端发送登录请求，携带 user_id
    if ($msg['type'] === 'login') {
        $user_id = $msg['user_id'];
        $connection->user_id = $user_id;
        $worker->uidConnections[$user_id] = $connection;
        echo "User {$user_id} logged in.\n";
    }

    // 广播消息给所有人
    if ($msg['type'] === 'broadcast') {
        $content = $msg['content'];
        foreach ($worker->connections as $conn) {
            $conn->send(json_encode(['type' => 'notice', 'content' => $content]));
        }
    }
};
    

5. 分布式环境下的挑战与解决方案

当应用扩展至多服务器或多 Worker 进程时，单机内存存储的连接列表失效。此时需引入中间件进行跨进程/跨节点通信。推荐架构如下：

graph TD
            A[Client] --> B{Load Balancer}
            B --> C[Server 1: Worker1]
            B --> D[Server 2: Worker2]
            C --> E[(Redis Pub/Sub)]
            D --> E
            E --> F[广播消息同步]
            C --> G[本地 connections]
            D --> H[本地 connections]

6. 基于 Redis 的跨进程广播机制

利用 Redis 的发布/订阅模式，实现多 Worker 间的消息同步：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 在 onWorkerStart 中监听频道
if ($worker->id === 0) { // 仅一个进程监听，避免重复
    $redis->subscribe(['broadcast_channel'], function($redis, $channel, $message) use ($worker) {
        foreach ($worker->connections as $conn) {
            $conn->send($message);
        }
    });
}

// 发送广播时，推送到 Redis
$redis->publish('broadcast_channel', json_encode(['type' => 'global', 'msg' => 'Hello All']));
    

7. 用户级精准推送与连接管理优化

为支持按用户 ID 推送，可构建更精细的连接注册机制：

1. 客户端连接后立即发送 token 或 user_id
2. 服务端验证合法性并绑定 connection->user_id
3. 将 user_id 写入 Redis Hash 表：HSET ws_connections user_id process_id:conn_id
4. 广播时通过 Lua 脚本获取所有在线用户
5. 通过 GatewayClient 或自定义协议转发至目标进程
6. 目标进程查找本地 connection 并 send
7. 定期清理过期连接（配合心跳机制）
8. 支持踢人下线、多端互斥登录等策略
9. 记录连接日志用于审计与追踪
10. 结合 Swoole 替代 Workerman 可进一步提升性能

8. 高可用设计建议

为保障生产环境稳定性，应考虑以下措施：

• 启用守护进程模式：'daemonize' => true
• 设置心跳检测间隔（如每 30 秒 ping 一次）
• 使用 Supervisor 管理 Worker 进程生命周期
• 集成日志系统（Monolog + ELK）
• 压力测试工具：wrk、artillery.io 模拟千人并发