Redis 的发布/订阅（Pub/Sub）机制

发表于 2026-06-30 更新于 2026-06- 30

作者 mdo

15~20 分钟 阅读

在 think-swoole 中，如果项目开启了多个 Worker 进程或者部署了多台服务器，直接遍历 server->connections 只能触及当前进程下的连接。

要实现跨进程、跨服务器的广播，最成熟的方案是引入 Redis 的发布/订阅（Pub/Sub） 机制。这类似于 Socket.io 官方的 Redis Adapter 原理。

以下是完整的架构设计与后端实现步骤：

核心架构原理

客户端连接：分散在不同的进程或服务器上，各自保持长连接。
发布消息：任何一个进程需要广播时，将消息 Publish（发布）到 Redis 的指定频道（Channel）。
订阅并分发：每台服务器在启动时，开启一个常驻进程 Subscribe（订阅）该频道。收到 Redis 消息后，遍历本地连接进行单机内广播。

具体代码实现

1. 配置常驻订阅进程 (Swoole 自定义进程)

由于 Redis 订阅（Subscribe）是阻塞的，必须将其运行在 Swoole 的自定义子进程中。

新建一个自定义进程类 app\process\RedisSubscribeProcess.php：

php

namespace app\process;

use think\swoole\Process;
use think\swoole\Manager;
use think\facade\Cache;

class RedisSubscribeProcess extends Process
{
    protected $manager;

    public function __construct(Manager $manager)
    {
        $this->manager = $manager;
        parent::__construct();
    }

    public function handle(): void
    {
        // 获取底层 swoole server 实例
        $server = $this->manager->getServer();
        
        // 获取 Redis 原生连接 (确保配置中驱动为 redis)
        $redis = Cache::store('redis')->handler();
        
        // 设置无超时限制
        $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);

        // 阻塞订阅 'ws_global_broadcast' 频道
        $redis->subscribe(['ws_global_broadcast'], function ($redis, $chan, $msg) use ($server) {
            // 收到 Redis 消息后，遍历当前服务器上的所有连接
            foreach ($server->connections as $fd) {
                if ($server->isEstablished($fd)) {
                    // 推送消息给客户端
                    $server->push($fd, $msg);
                }
            }
        });
    }
}

请谨慎使用此类代码。

2. 在 config/swoole.php 中注册进程

打开 config/swoole.php，将自定义进程配置到 websocket.processes 或 hot_update.include 相关的配置项中（或直接在 queue 同级的 processes 数组中）：

php

return [
    // ... 其他配置
    'websocket' => [
        'enable'        => true,
        // ...
    ],
    // 注册常驻进程
    'processes' => [
        \app\process\RedisSubscribeProcess::class,
    ],
];

请谨慎使用此类代码。

3. 业务层发布广播

现在，在控制器或任何业务逻辑中，广播消息不再需要遍历 fd，只需向 Redis 发布一条消息即可。

php

namespace app\controller;

use think\facade\Cache;

class Broadcast
{
    public function send()
    {
        $data = [
            'event'   => 'message',
            'content' => '这是一条跨进程、跨服务器的全局广播消息！',
            'time'    => date('Y-m-d H:i:s')
        ];

        // 获取 Redis 实例并发布消息
        $redis = Cache::store('redis')->handler();
        $redis->publish('ws_global_broadcast', json_encode($data));

        return json(['status' => 'success', 'msg' => '跨服广播已发布']);
    }
}

如果在你的 config/swoole.php 中没有默认提供 processes 配置项，这通常是因为不同版本的 think-swoole（例如 v3.x 或 v4.x）默认生成的配置文件模版有所精简，或者该版本改由事件监听的形式来挂载自定义进程。[1]

你可以通过以下两种最有效的方法来解决这个问题：

方法一：直接在 config/swoole.php 中手动添加（推荐）

think-swoole 的底层容器其实支持加载该配置。如果文件中没有，你可以直接手动写进去。

打开 config/swoole.php，在返回的数组最外层，直接手动补上 'processes' 数组：

php

return [
    'server' => [
        'host' => '0.0.0.0',
        'port' => 9501,
        // 其他基础配置...
    ],

    // ===== 手动加上这一段 =====
    'processes' => [
        \app\process\RedisSubscribeProcess::class, 
    ],
    // =========================

    'websocket' => [
        'enable' => true,
        // 其他ws配置...
    ],
];

请谨慎使用此类代码。

添加后，重启 Swoole 服务 (php think swoole restart)。在 Linux 终端中运行 ps -ef | grep swoole，如果能看到比平时多出一个子进程，说明挂载成功。

方法二：通过 Swoole 的 swoole.init 事件动态注入

如果手动添加配置项后，系统提示配置未生效，说明你使用的 think-swoole 版本完全移除了该配置。此时，可以通过 ThinkPHP 的事件系统，在 Swoole 初始化时动态将进程加入 Server。

1. 创建事件监听类

通过命令行生成一个专门用于初始化 Swoole 的监听器：

bash

php think make:listener SwooleInit

请谨慎使用此类代码。

2. 在监听器中手动绑定自定义进程

打开 app/listener/SwooleInit.php，修改 handle 方法。利用 Swoole 原生的 $server->addProcess() 动态挂载：

php

namespace app\listener;

use Swoole\Process as SwooleProcess;
use think\swoole\Manager;
use think\facade\Cache;

class SwooleInit
{
    protected $manager;

    public function __construct(Manager $manager)
    {
        $this->manager = $manager;
    }

    public function handle($event)
    {
        // 获取 Swoole Server 实例
        $server = $this->manager->getServer();

        // 创建一个 Swoole 原生自定义进程
        $process = new SwooleProcess(function ($process) use ($server) {
            // 设置进程名称（方便查看）
            $process->name('think-swoole: redis-subscribe');

            // 引入 Redis 阻塞订阅逻辑
            $redis = Cache::store('redis')->handler();
            $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);

            // 开始订阅
            $redis->subscribe(['ws_global_broadcast'], function ($redis, $chan, $msg) use ($server) {
                // 收到消息后，全服广播
                foreach ($server->connections as $fd) {
                    if ($server->isEstablished($fd)) {
                        $server->push($fd, $msg);
                    }
                }
            });
        }, false, 0, true); // 启用协程支持

        // 将自定义进程添加到 Swoole Server 中
        $server->addProcess($process);
    }
}

请谨慎使用此类代码。

3. 绑定事件

打开全局事件配置文件 app/event.php，在 listen 数组中，将上面的监听类绑定到 swoole.init 事件上：

php

return [
    'bind'    => [],
    'listen'  => [
        // 监听 think-swoole 的初始化事件
        'swoole.init' => [
            \app\listener\SwooleInit::class,
        ],
    ],
    'subscribe' => [],
];

请谨慎使用此类代码。

使用方法二的好处在于，它完全绕过了 think-swoole 自身的配置限制，直接作用于底层的 Swoole 实例，兼容性极高。

技术

许可协议: CC BY 4.0