README

ThinkPHP8 的 Workerman / GatewayWorker 常驻服务集成包——一条命令启动并管理 WebSocket 网关、TCP 服务、队列消费。

特性

一条命令管理所有常驻服务：php think workers [start|stop|restart|reload|status]
GatewayWorker 体系：Register 注册中心 + BusinessWorker 业务进程 + 多 Gateway 网关，多个 WebSocket 入口共享同一套业务进程
多进程无连接隔离：内置 Gateway API（sendToUid / sendToGroup / sendToAll / bindUid / joinGroup），天然支持多进程、分布式部署
TCP 自定义协议服务：如物联网设备接入，业务继承 Handler 即可
队列常驻消费：Workerman 进程内运行 think-queue 的 daemon 循环，相当于常驻内存版 php think queue:work --daemon
开发热更新：前台启动自动监控 app / config 下 .php 变化，自动重载业务进程且客户端连接不断
三层开关：worker.enable 总开关 → 分组 enable → 项级 enable，任意粒度控制服务启停
服务上下文：每次启动的服务清单（类型 / 监听 / 进程数 / 处理器）写入 runtime/worker/services.json
零配置文件复制：主项目无 config/matrix_worker.php 时自动注入包内模板

环境要求

依赖	版本
PHP	>= 8.3
topthink/framework	^8.1
workerman/gateway-worker	^4.0

安装

composer require adminmatrix/matrix_worker

快速开始

1. 启动服务

php think workers          # 前台启动（调试）
php think workers start -d # 守护进程启动

启动后默认注册 6 个服务：

Register          text://127.0.0.1:1236      注册中心
business          none                       业务进程 -> Events
worker            websocket://0.0.0.0:2348   主网关（内部通讯 2000）
gateway           websocket://0.0.0.0:2347   组级默认网关（内部通讯 2100）
测试聊天服务       websocket://0.0.0.0:2346   额外网关（内部通讯 2200）
默认队列          none                       队列 default 消费

2. 前端连接

const ws = new WebSocket('ws://127.0.0.1:2346');
ws.onmessage = e => {
    const data = JSON.parse(e.data);
    if (data.type === 'ping') { ws.send('{"type":"pong"}'); return; } // 心跳回应（配置 pingInterval 后）
    console.log(data);
};
ws.onopen = () => ws.send('hello');
// 收到：{"type":"welcome","msg":"连接成功","client_id":"7f000001..."}
// 收到：{"type":"echo","msg":"hello"}

命令

命令	说明
`php think workers`	前台启动（调试模式）
`php think workers start -d`	守护进程启动
`php think workers stop`	停止服务
`php think workers restart`	重启服务
`php think workers reload`	平滑重启（重载业务代码）
`php think workers status`	查看进程状态

配置

主项目无配置时自动使用包内模板；需要自定义时复制到主项目 config/matrix_worker.php。

配置优先级：workers 项 > 分组 > 全局 worker。

return [
    // 开发模式热更新（仅前台启动生效，守护进程 start -d 自动关闭）
    'monitor' => [
        'enable'   => true,               // 开关：false 时前台启动也不监控
        'interval' => 1,                  // 扫描间隔秒
        'paths'    => ['app', 'config'],  // 监控目录（相对主项目根目录）
    ],
    // 队列消费服务（不监听端口，常驻消费 think-queue）
    'queue' => [
        'enable' => true,
        'queue'  => 'default',      // 分组默认：监听的队列名
        'sleep'  => 3,              // 分组默认：空闲休眠秒数
        'count'  => 1,
        'workers' => [
            [
                'enable' => true,
                'name'   => '默认队列',
                'queue'  => 'default',   // 项缺省时用分组默认
            ],
        ],
    ],
    // 全局总开关 + 全局默认值（worker.enable = false 时所有服务都不注册）
    'worker' => [
        'enable' => true,
        'host'   => '0.0.0.0',
        'port'   => 2348,            // 默认主网关端口
        'count'  => 1,
    ],
    // GatewayWorker 体系（Register 注册中心 + BusinessWorker 业务进程 + 多 Gateway 网关）
    'gateway' => [
        'enable' => true,
        // Register 注册中心（Gateway / BusinessWorker 通过它互相发现）
        'register' => [
            'host' => '127.0.0.1',
            'port' => 1236,
        ],
        // BusinessWorker 业务进程（eventHandler 静态方法类，业务逻辑写这里）
        'business' => [
            'name'    => 'business',
            'count'   => 1,
            'handler' => 'adminmatrix\worker\websocket\Events',
        ],
        // 网关内部通讯
        'lanIp'     => '127.0.0.1',      // 内部通讯 IP（多机部署填内网 IP）
        'startPort' => 2000,             // 内部通讯起始端口（多网关自动按 100 错开）
        // 组级默认网关（始终注册）
        'name'      => 'gateway',
        'host'      => '0.0.0.0',
        'port'      => 2347,
        'count'     => 1,
        // 额外网关（多个 ws 入口，共享同一套 BusinessWorker 业务进程，
        // 业务逻辑统一写在 business.handler 指向的 Events 类，见「业务开发」）
        'workers' => [
            [
                'enable' => true,
                'name'   => '测试聊天服务',
                'host'   => '0.0.0.0',
                'port'   => 2346,
            ],
            // 完整可用字段（按需打开）：
            // [
            //     'enable'               => true,                  // 项级开关（缺省视为开启）
            //     'name'                 => '客服服务',
            //     'host'                 => '0.0.0.0',
            //     'port'                 => 2350,
            //     'count'                => 1,                     // 进程数
            //     'handler'              => 'app\worker\Events',   // 独立业务事件类（缺省用 business.handler）
            //     'lanIp'                => '127.0.0.1',           // 内部通讯 IP（多机部署填内网 IP）
            //     'startPort'            => 2300,                  // 内部通讯起始端口（缺省自动按 100 错开）
            //     'pingInterval'         => 30,                    // 心跳间隔秒（> 0 开启心跳，0 不心跳）
            //     'pingNotResponseLimit' => 2,                     // 连续 N 个周期无上行消息则断开（0 只发不踢）
            //     'pingData'             => '{"type":"ping"}',     // 服务端下发的心跳包内容
            // ],
        ],
    ],
    // TCP 服务（多实例，自定义协议如物联网设备接入）
    'tcp' => [
        'enable' => false,
        'host'   => '0.0.0.0',
        'port'   => 2349,
        'count'  => 1,
        'workers' => [
            // 完整可用字段（按需打开，handler 继承 adminmatrix\worker\tcp\Handler）：
            // [
            //     'enable'  => true,                              // 项级开关（缺省视为开启）
            //     'name'    => '设备接入',
            //     'host'    => '0.0.0.0',
            //     'port'    => 2351,
            //     'count'   => 1,                                 // 进程数
            //     'handler' => 'app\worker\tcp\DeviceHandler',     // 业务处理器
            // ],
        ],
    ],
];

开关层级

worker.enable = false        # 所有服务都不注册
├── gateway.enable = false   # GatewayWorker 体系整体不注册
│   └── workers[].enable     # 单个网关不注册
├── tcp.enable = false       # TCP 服务整体不注册
│   └── workers[].enable
└── queue.enable = false     # 队列消费整体不注册
    └── workers[].enable

业务开发

WebSocket 业务（Events 静态方法类）

业务逻辑写在 gateway.business.handler 指向的类里，静态方法，拿到的是 client_id（不是 connection 对象），通过 Gateway API 操作连接：

namespace app\worker;

use GatewayWorker\Lib\Gateway;

class Events
{
    public static function onConnect($client_id): void
    {
        Gateway::sendToClient($client_id, json_encode([
            'type'      => 'welcome',
            'msg'       => '连接成功',
            'client_id' => $client_id,
        ], JSON_UNESCAPED_UNICODE));
    }

    public static function onMessage($client_id, $message): void
    {
        $data = json_decode($message, true);
        switch ($data['type'] ?? '') {
            case 'bind':   // 绑定用户（多端登录自动合并）
                Gateway::bindUid($client_id, $data['uid']);
                break;
            case 'join':   // 加入房间
                Gateway::joinGroup($client_id, $data['room']);
                break;
            case 'say':    // 房间消息
                Gateway::sendToGroup($data['room'], $data['msg']);
                break;
            case 'all':    // 全局广播
                Gateway::sendToAll($data['msg']);
                break;
        }
    }

    public static function onClose($client_id): void {}
}

指向业务类：

'business' => [
    'handler' => 'app\worker\Events',
],

多网关不同业务（workers 项级 handler）

每个网关项可配独立 handler（业务事件类），包会为它创建专属 BusinessWorker 进程并固定路由——各网关业务完全隔离；不配 handler 的网关走默认 business.handler。多个网关配同一个 handler 时共享同一个业务进程：

'workers' => [
    [   // 不配 handler -> 走默认 business.handler（adminmatrix\worker\websocket\Events）
        'name' => '测试聊天服务',
        'port' => 2346,
    ],
    [   // 独立业务事件类 -> 专属 BusinessWorker 进程
        'name'    => '客服服务',
        'port'    => 2350,
        'handler' => 'app\worker\CustomerEvents',
    ],
],

CustomerEvents 与默认 Events 写法相同（静态方法 + Gateway API）：

namespace app\worker;

use GatewayWorker\Lib\Gateway;

class CustomerEvents
{
    public static function onConnect($client_id): void
    {
        Gateway::sendToClient($client_id, json_encode([
            'type' => 'welcome', 'msg' => '客服服务连接成功',
        ], JSON_UNESCAPED_UNICODE));
    }

    public static function onMessage($client_id, $message): void
    {
        // 客服业务...
    }

    public static function onClose($client_id): void {}
}

心跳配置

网关项（或 gateway 组级）配置心跳，自动踢掉死连接：

'pingInterval'         => 30,                 // 每 30 秒向客户端下发一次 pingData
'pingNotResponseLimit' => 2,                  // 连续 2 个周期（60 秒）无上行消息则断开
'pingData'             => '{"type":"ping"}',  // 下发的心跳包内容

前端收到心跳包后回一条任意消息即可保活（推荐 {"type":"pong"}），见「前端连接」示例。

热更新（开发模式）

前台启动（php think workers）时自动开启文件监控，改 app / config 下的 .php 文件后约 1 秒自动生效，无需手动 reload，客户端连接不断：

[monitor] 检测到文件变化，平滑重载业务进程（连接保持）...

原理：Gateway（持有客户端连接）和 Register 不参与 reload；reload 信号只重启业务进程（BusinessWorker / TCP）重新加载业务代码，客户端连接保持。Queue 消费进程不参与 reload（think-queue 的 daemon 循环不处理重载信号，队列代码变更请用 restart）。守护进程模式（start -d）自动关闭监控；监控目录 / 间隔见 monitor 配置。

注意：修改 config / Worker 启动逻辑相关代码仍需 php think workers restart（reload 只热更新业务代码）。

常用 Gateway API：

方法	说明
`Gateway::sendToClient($client_id, $msg)`	发给单个连接
`Gateway::sendToUid($uid, $msg)`	发给用户（`bindUid` 后，多端全收）
`Gateway::sendToGroup($group, $msg)`	发给分组（`joinGroup` 后）
`Gateway::sendToAll($msg)`	全局广播
`Gateway::bindUid($client_id, $uid)`	连接绑定用户
`Gateway::joinGroup($client_id, $group)`	连接加入分组
`Gateway::closeClient($client_id)`	踢掉连接

TCP 业务（Handler 继承）

namespace app\worker\tcp;

use Workerman\Connection\TcpConnection;
use adminmatrix\worker\Handler;

class Device extends Handler
{
    public function onConnect(TcpConnection $connection): void
    {
        // 设备上线
    }

    public function onMessage(TcpConnection $connection, string $data): void
    {
        // TCP 收到的是原始数据帧，业务层自行拆包
        $connection->send('recv: ' . $data);
    }

    public function onClose(TcpConnection $connection): void
    {
        // 设备离线
    }
}

配置指向业务类：

'tcp' => [
    'enable' => true,
    'workers' => [
        ['name' => '设备接入', 'host' => '0.0.0.0', 'port' => 2349, 'handler' => 'app\worker\tcp\Device'],
    ],
],

目录结构

src/
├── command/Run.php               # think 命令入口（php think workers）
├── config/matrix_worker.php      # 配置模板（主项目无配置时自动注入）
├── HandlerInterface.php          # 公共契约：必须实现的 4 个连接事件 + 可选事件标注
├── Handler.php                   # 公共基类：集成 think console Output（info/error 等日志输出）
├── Service.php                   # think 服务注册（配置注入）
├── Worker.php                    # 启动器 + 服务容器基类（服务列表 / think Table 渲染 / 上下文记录）
├── server/                       # 服务容器
│   ├── Register.php              # GatewayWorker 注册中心
│   ├── Business.php              # GatewayWorker 业务进程
│   ├── Gateway.php               # GatewayWorker 网关（含心跳配置）
│   ├── WebSocket.php             # WebSocket 直连容器（备用）
│   ├── Tcp.php                   # TCP 容器
│   └── Queue.php                 # 队列消费容器
├── websocket/
│   ├── Events.php                # Gateway 默认业务事件类
│   ├── HandlerInterface.php      # WebSocket 契约（+ onWebSocketConnect 握手鉴权）
│   └── Handler.php               # 直连模式默认处理器（备用）
└── tcp/
    ├── HandlerInterface.php      # TCP 契约（原始帧 / 粘包拆包标注）
    └── Handler.php               # TCP 默认处理器

多机部署

所有机器的 register.host 指向同一台 Register 机器
每台机器的 lanIp 填自己的内网 IP
各机器网关 startPort 不重叠（同机多网关会自动按 100 错开）

License

MIT

adminmatrix / matrix_worker

Package info

Security

Statistics