# matrix_worker **Repository Path**: adminmatrix/matrix_worker ## Basic Information - **Project Name**: matrix_worker - **Description**: No description available - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-09-16 - **Last Updated**: 2026-09-16 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # matrix_worker 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 | ## 安装 ```bash composer require adminmatrix/matrix_worker ``` ## 快速开始 ### 1. 启动服务 ```bash 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. 前端连接 ```js 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**。 ```php 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 操作连接： ```php 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 {} } ``` 指向业务类： ```php 'business' => [ 'handler' => 'app\worker\Events', ], ``` ### 多网关不同业务（workers 项级 handler） 每个网关项可配独立 `handler`（业务事件类），包会为它创建专属 BusinessWorker 进程并固定路由——各网关业务完全隔离；不配 `handler` 的网关走默认 `business.handler`。多个网关配同一个 handler 时共享同一个业务进程： ```php 'workers' => [ [ // 不配 handler -> 走默认 business.handler（adminmatrix\worker\websocket\Events） 'name' => '测试聊天服务', 'port' => 2346, ], [ // 独立业务事件类 -> 专属 BusinessWorker 进程 'name' => '客服服务', 'port' => 2350, 'handler' => 'app\worker\CustomerEvents', ], ], ``` `CustomerEvents` 与默认 Events 写法相同（静态方法 + Gateway API）： ```php 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 组级）配置心跳，自动踢掉死连接： ```php '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 继承） ```php 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 { // 设备离线 } } ``` 配置指向业务类： ```php '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](LICENSE)