Download the PHP package easyswoole/nsq without Composer

EasySwoole NSQ 2.x

基于 Swoole\Coroutine\Socket 的 NSQ V2 客户端。运行时只依赖 PHP >= 8.1 和 Swoole >= 5.1；消息时间戳使用纳秒整数，要求 64 位 PHP。2.x 使用全新 API，不兼容 1.x。

支持普通发布、批量发布、延迟发布、订阅消费、手动确认、心跳、AUTH、TLS / mTLS 和 HTTP(S) 服务发现。消息体按二进制字符串处理，无需 EasySwoole 框架即可使用。

阅读导航

• 安装与测试：环境要求、PHPUnit、本地 mock 服务。
• 发布消息：PUB、MPUB、DPUB 及发布失败的含义。
• 消费消息：自动确认、手动确认、并发消费和停止。
• 配置与认证：参数默认值、TLS、客户端证书和心跳。
• 服务发现：查询 lookupd 并连接发现的 nsqd。
• 异常与重连：错误类型、重投和重连边界。
• API 与源码导读：常用方法和关键代码位置。

安装与测试

当前仓库为 2.x 开发分支，直接在仓库目录安装依赖即可。作为其他项目的依赖时，使用仓库实际可用的 2.x 版本约束；正式发布 2.x 后可使用 composer require easyswoole/nsq:^2.0。

开发时安装依赖后执行：

沿用项目的 PHPUnit ^13.4，运行测试需要 PHP >= 8.4.1。库本身的最低运行版本仍为 PHP 8.1。

可用 php -v、php --ri swoole 检查 CLI 环境。开发依赖要求较高的 PHP 版本，不改变库本身的最低运行版本；部署应用且不需要其开发依赖时，使用 composer install --no-dev。

tests 包含独立的协议/状态机单元测试，以及自动启动本地模拟服务端的 Socket、HTTP、TLS 测试，无需部署 NSQ。SocketTransportTest / LookupTest 的本地端口不可用时会标记 skipped；mock nsqd 和 TLS 集成测试在启动失败时直接报错。在允许本地网络的 CI 中，使用以下命令要求前两类测试也不允许跳过：

按问题定位时，可以只运行相关测试：

TLS 测试中的不可信 CA、证书名称不匹配是预期的拒绝场景，Swoole 可能输出相应诊断日志；以 PHPUnit 最终的失败、错误和跳过统计判断测试结果。

本地 mock nsqd

启动有消息队列状态的 NSQ V2 模拟服务（仅监听 localhost）：

模拟服务使用独立的协议解析实现，支持 IDENTIFY、AUTH、PUB、MPUB、DPUB、SUB、RDY、FIN、REQ、TOUCH、NOP、CLS，以及 channel 分发、心跳、处理超时与断线后的重投。启动后把下方示例的 Config 指向 127.0.0.1:4150 即可试用；终端按 Ctrl+C 停止，内存中的消息随进程退出而丢失。

该命令启动的是明文 mock；TLS 使用 tests/fixtures/tls/mock-nsqd.php 作为独立进程测试夹具，由 TlsIntegrationTest 自动配置并启动。mock 不模拟压缩、磁盘持久化和完整集群行为，不能替代针对真实 nsqd 的兼容性测试。

MockNsqdIntegrationTest 自动启动该服务，在真实 TCP 连接上验证发布消费、批量发布、REQ 延迟和重试次数、DPUB、心跳 NOP、多 channel 分发及认证。运行这些测试无需手动启动服务：

发布消息

所有网络操作均需在 Swoole 协程内执行；构造客户端不会建立连接，首次操作时自动连接并完成 IDENTIFY/AUTH。

消息体为非空二进制字符串，JSON 等应用层编码由调用方处理。MPUB 必须包含至少一条消息。成功发布以服务端 OK 为准；超时或断线不会自动重试，避免因服务端已接收消息而重复发布。调用方应结合业务幂等要求决定是否重试。

topic 和 channel 名称只允许字母、数字、.、_、-，可追加 #ephemeral；总长度为 1～64 字节，包含该后缀。deferredPublish() 的延迟单位为毫秒，实际允许上限由 nsqd 配置决定。

可以显式调用 $nsq->connect() 提前完成握手；重复调用不会重新建连。isConnected() 返回本地维护的连接状态，不主动探测远端是否仍在线，空闲连接断开通常在下一次读写时被发现。

消费消息

consume() 串行处理，每次发送 RDY 1。回调只有严格返回 false 时自动 REQ，其他返回值（含未显式返回时的 null）自动 FIN。回调可手动 finish() 或 requeue()，客户端不会重复确认。

手动接收与确认

需要自行决定确认时机时，不调用 consume()，使用以下流程：

subscribe() 成功后 RDY 初始为 0。ready($count) 设置服务端剩余投递额度，不是在原额度上累加；每次投递会扣减，手动模式需要自行补充。FIN / REQ / TOUCH 不等待成功响应，服务端可能在后续读取时返回错误帧。

只能操作本连接当前仍在处理的原始 Message 对象，不能自行构造、克隆或使用已确认/旧连接的对象。消息字段如下：

并发消费与停止

同一 topic/channel 下的多个消费者分担消息；不同 channel 分别接收该 topic 的消息。每个客户端由一个协程独占，并发消费时在各协程内部创建客户端：

订阅后不能使用同一连接发布消息。回调内调用 $client->stop() 时，当前消息仍按回调结果确认，随后退出循环；连接有效时发送 RDY 0 暂停新投递，连接本身保持打开。stop() 只改变循环标记，不会唤醒正阻塞的 receive()。

close() 立即关闭连接并清除本地订阅和消息状态，不发送 CLS，也不会替未处理消息发送 FIN。未确认消息由服务端重新投递，因此业务处理应考虑重复消息和幂等性。读取失败后需重新 subscribe()，旧 Message 对象不能继续确认。

配置与认证

Config 为只读配置对象，构造时校验参数，首次网络操作时才使用配置建立连接：

超时单位为秒；-1 表示无限等待。maxFrameSize 限制收到的帧大小，包含 4 字节帧类型，不包含 4 字节长度前缀。只有服务端声明 auth_required 时才发送 AUTH。当前实现不支持 Snappy、Deflate，启用这些 IDENTIFY 选项会直接报错。

TLS 与双向认证

TLS 配置沿用 EasySwoole Consul 的字段和校验规则。tls 默认为 null（明文）；传入 new TlsConfig() 即要求 TLS，默认 verifyPeer: true、allowSelfSigned: false，使用系统 CA 校验证书。serverName 默认使用连接的 host，用于 SNI 和服务器名称校验；连接 IP 地址而证书签发给域名时，显式填写证书域名。

NSQ 使用同一 TCP 连接升级 TLS：明文 MAGIC / IDENTIFY → 服务端确认 tls_v1: true → TLS 握手 → 加密的 OK → AUTH / 发布 / 订阅。TLS 配置会自动生成 IDENTIFY 的 tls_v1，不要手动设置冲突值。服务端未协商 TLS、证书校验或握手失败时关闭连接，不会退回明文；AUTH 秘钥只在成功升级并收到加密 OK 后发送。TLS 握手使用 connectTimeout。

自定义 CA 支持 caFile 和 OpenSSL 哈希证书目录 caPath。客户端证书认证配置如下：

clientCertificate 和 clientPrivateKey 必须同时配置；passphrase 必须有私钥；allowSelfSigned: true 要求保留 verifyPeer: true。显式配置 verifyPeer: false 可关闭校验，默认始终开启。Swoole 需要编译 OpenSSL 支持（旧版本使用 --enable-openssl）。

nsqd 服务端使用 --tls-cert / --tls-key 配置证书，--tls-required=true 强制客户端升级；双向认证使用 --tls-client-auth-policy=require-verify --tls-root-ca-file=/etc/nsq/tls/ca.pem。TLS 不会隐藏升级前的 IDENTIFY 元数据。

TlsIntegrationTest 自动生成临时测试 CA、服务端和客户端证书，使用独立 PHP 进程模拟 NSQ 的 TLS 升级，验证真实加密发布/消费、AUTH、mTLS、口令私钥、CA 目录、证书/名称拒绝、握手超时及 HTTPS 服务发现。测试结束删除临时证书，无需仓库内保存私钥。

心跳与长任务

客户端在等待握手、发布响应或消息时处理心跳并回复 NOP，不会启动后台读协程。长时间空闲的发布连接应及时 close()，或者设置 identify: ['heartbeat_interval' => -1]。readTimeout: -1 只表示无限等待数据，不会禁用服务端心跳。

回调执行期间不会并行读心跳；消费回调应在心跳和消息超时前完成。长任务可在处理步骤之间主动 touch($message) 延长消息处理时间，并调整 IDENTIFY 的 heartbeat_interval / msg_timeout 或服务端配置。TOUCH 不会代替客户端读取和响应心跳。

服务发现

发现使用 Swoole 协程 HTTP 客户端，合并去重所有成功节点的结果；只有全部 lookupd 请求失败时才抛出 LookupException。结果为空是合法状态。发现后的连接创建、负载分配、周期刷新和重连策略由业务方管理。

HTTPS 服务发现默认校验证书，也可复用同一 TlsConfig：

显式指定 lookup 的 tls 时，所有地址必须为 HTTPS。

lookupd 返回的是 nsqd TCP 地址，不是可直接传给 host 的完整 URL。需要按地址解析主机和端口，再创建客户端：

lookupd 的 HTTPS 配置只保护发现请求；连接发现的 nsqd 是否启用 TLS，由各自的 Config::tls 决定。上述示例消费明文 nsqd，TLS 部署时需另外传入对应的 TlsConfig。

异常与重连

握手 JSON 非法时也可能抛出 JsonException；consume() 的业务回调异常按前文规则向上传递。不要把业务异常全部当作网络异常无限重试。

客户端不会自动重放失败命令或自动恢复订阅。发布断线时，“未收到 OK”不等于“消息一定未发布”，应先考虑业务幂等，再决定是否重新调用发布方法；未订阅的客户端在下一次发布时可重新建连。消费断线时，应重新 subscribe()，然后再次 consume() 或 ready() / receive()；根据业务需要加入退避、重试上限和日志。

consume() 不提供最大重试次数、死信队列或消息去重；应用可读取 Message::attempts 决定后续处理。stop()、close() 不等价于服务端确认，处理成功必须通过 FIN 表达。

API 与源码导读

核心代码按职责分层，建议从入口向协议和 Socket 层阅读：

当前不提供连接池、自动负载均衡、自动重连、定时刷新发现结果或压缩功能，这些策略由调用方按部署需求实现。

协议参考：NSQ TCP V2、nsqlookupd HTTP 接口。