README

基于 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。

开发时安装依赖后执行：

composer install
composer test

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

场景	环境要求
使用客户端	64 位 PHP >= 8.1、Swoole >= 5.1，在协程内执行网络操作
使用 TLS / HTTPS	Swoole 已编译 OpenSSL 支持；证书策略见下文
开发及运行测试	PHP >= 8.4.1，安装 Composer 开发依赖
完整 Socket / TLS 测试	允许监听本地端口和启动 PHP 子进程；PHP OpenSSL 扩展用于生成临时测试证书

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

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

NSQ_REQUIRE_SOCKET_TESTS=1 composer test

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

vendor/bin/phpunit --filter PacketTest
vendor/bin/phpunit --filter MockNsqdIntegrationTest
vendor/bin/phpunit --filter TlsIntegrationTest

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

本地 mock nsqd

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

php tests/mock-nsqd.php 4150
# 可选认证：MOCK_NSQ_AUTH_SECRET=secret php tests/mock-nsqd.php 4150

模拟服务使用独立的协议解析实现，支持 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 分发及认证。运行这些测试无需手动启动服务：

vendor/bin/phpunit --filter MockNsqd

发布消息

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

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Nsq;

require __DIR__ . '/vendor/autoload.php';

Swoole\Coroutine\run(function (): void {
    $nsq = new Nsq(new Config(host: '127.0.0.1', port: 4150));
    try {
        $nsq->publish('topic.tests', 'hello');
        $nsq->multiPublish('topic.tests', ['first', 'second']);
        $nsq->deferredPublish('topic.tests', 'later', 1000); // 毫秒
    } finally {
        $nsq->close();
    }
});

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

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

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

消费消息

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Message\Message;
use EasySwoole\Nsq\Nsq;

require __DIR__ . '/vendor/autoload.php';

Swoole\Coroutine\run(function (): void {
    $nsq = new Nsq(new Config(readTimeout: -1));
    try {
        $nsq->subscribe('topic.tests', 'tests.consuming');
        $nsq->consume(function (Message $message, Nsq $client): bool {
            echo $message->body;
            // $message->id、attempts、timestamp（Unix 纳秒）均为只读字段。
            // $client->touch($message);  // 延长消息处理时限
            // $client->stop();           // 当前回调完成并确认后退出
            return true;                 // true/null => FIN；false => REQ
        }, requeueDelayMilliseconds: 1000);
    } finally {
        $nsq->close();
    }
});

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

回调结果	客户端行为
返回 `true` / `null`，或其他非 `false` 值	若消息尚未手动确认，发送 FIN
严格返回 `false`	若消息尚未手动确认，按 `requeueDelayMilliseconds` 发送 REQ
抛出异常	若连接仍有效且消息未确认，先尝试 REQ，再把原异常抛出并退出循环
处理异常后 REQ 也失败	抛出新的 `RuntimeException`，原回调异常保存在 `getPrevious()`
回调手动 FIN / REQ	按手动操作处理，不再自动确认

手动接收与确认

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

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Nsq;

require __DIR__ . '/vendor/autoload.php';

Swoole\Coroutine\run(function (): void {
    $consumer = new Nsq(new Config(readTimeout: -1));
    try {
        $consumer->subscribe('topic.tests', 'manual.consuming');
        $consumer->ready(1);
        $message = $consumer->receive();

        echo $message->body;
        $consumer->finish($message);
        // 如果业务处理失败，改为 $consumer->requeue($message, 1000)。
        // 处理较长任务时可调用 touch($message) 延长服务端消息超时。
    } finally {
        $consumer->close();
    }
});

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

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

字段	类型	含义
`id`	`string`	16 字节消息 ID
`body`	`string`	原始二进制正文，客户端不自动解码 JSON
`attempts`	`int`	服务端投递次数，可用于限制业务重试
`timestamp`	`int`	Unix 纳秒时间戳；不是秒，也不是毫秒

并发消费与停止

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

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Message\Message;
use EasySwoole\Nsq\Nsq;

require __DIR__ . '/vendor/autoload.php';

Swoole\Coroutine\run(function (): void {
    for ($worker = 0; $worker < 4; ++$worker) {
        Swoole\Coroutine::create(function () use ($worker): void {
            $consumer = new Nsq(new Config(readTimeout: -1));
            try {
                $consumer->subscribe('topic.tests', 'tests.consuming');
                $consumer->consume(function (Message $message) use ($worker): void {
                    echo "worker={$worker}: {$message->body}\n";
                });
            } finally {
                $consumer->close();
            }
        });
    }
});

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

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

配置与认证

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

参数	默认值	说明
`host`	`127.0.0.1`	nsqd TCP 主机，IPv6 地址填写裸地址，如 `::1`
`port`	`4150`	nsqd TCP 端口，不是 HTTP 或 lookupd 端口
`connectTimeout`	`3.0`	TCP 连接及 TLS 握手使用的超时，单位秒
`readTimeout`	`60.0`	定长读取超时，单位秒；需要长期等待可设为 `-1`
`writeTimeout`	`3.0`	完整发送命令的超时，单位秒
`maxFrameSize`	`16_777_216`	接收帧大小上限，含帧类型，不含长度前缀
`identify`	`[]`	客户端标识、心跳、消息超时等 NSQ 元数据
`authSecret`	`null`	服务端要求 AUTH 时使用的认证密钥
`tls`	`null`	TLS 配置，`null` 表示使用明文连接

$config = new EasySwoole\Nsq\Config(
    host: '127.0.0.1',
    port: 4150,
    connectTimeout: 3.0,
    readTimeout: 60.0,
    writeTimeout: 3.0,
    maxFrameSize: 16_777_216,
    identify: ['client_id' => 'worker-1', 'heartbeat_interval' => 30000],
    authSecret: 'secret',
    tls: new EasySwoole\Nsq\TlsConfig(
        caFile: '/etc/nsq/tls/ca.pem',
        serverName: 'nsqd.example.com',
    ),
);

超时单位为秒；-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。

`TlsConfig` 参数	默认值	说明
`verifyPeer`	`true`	校验服务端证书及服务器名称
`allowSelfSigned`	`false`	是否允许自签名证书，要求 `verifyPeer=true`
`caFile` / `caPath`	`null`	CA 文件或 OpenSSL 哈希证书目录，未指定时使用底层默认信任来源
`clientCertificate` / `clientPrivateKey`	`null`	双向 TLS 的客户端证书与私钥，必须成对配置
`passphrase`	`null`	客户端私钥口令
`serverName`	`null`	SNI 和证书名称校验使用的名称，默认取 `Config::host`

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

$tls = new EasySwoole\Nsq\TlsConfig(
    caFile: '/etc/nsq/tls/ca.pem',
    clientCertificate: '/etc/nsq/tls/client.pem',
    clientPrivateKey: '/etc/nsq/tls/client-key.pem',
    passphrase: 'private-key-password', // 无口令私钥可省略
    serverName: 'nsqd.example.com',
);
$config = new EasySwoole\Nsq\Config(tls: $tls);

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\Coroutine\run(function (): void {
    $lookup = new EasySwoole\Nsq\Lookup\Nsqlookupd([
        'http://127.0.0.1:4161',
        'http://127.0.0.2:4161',
    ]);
    $endpoints = $lookup->lookupHosts('topic.tests');
    // ['127.0.0.1:4150', '[::1]:4150', ...]
});

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

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

$lookup = new EasySwoole\Nsq\Lookup\Nsqlookupd(
    ['https://nsqlookupd.example.com:4162'],
    tls: new EasySwoole\Nsq\TlsConfig(caFile: '/etc/nsq/tls/ca.pem'),
);

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

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

use EasySwoole\Nsq\Config;
use EasySwoole\Nsq\Lookup\Nsqlookupd;
use EasySwoole\Nsq\Nsq;

require __DIR__ . '/vendor/autoload.php';

Swoole\Coroutine\run(function (): void {
    $lookup = new Nsqlookupd(['http://127.0.0.1:4161']);
    foreach ($lookup->lookupHosts('topic.tests') as $endpoint) {
        Swoole\Coroutine::create(function () use ($endpoint): void {
            $address = parse_url('tcp://' . $endpoint);
            $consumer = new Nsq(new Config(
                host: trim($address['host'], '[]'),
                port: $address['port'],
                readTimeout: -1,
            ));
            try {
                $consumer->subscribe('topic.tests', 'tests.consuming');
                $consumer->consume(function ($message): void { echo $message->body; });
            } finally {
                $consumer->close();
            }
        });
    }
});

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

异常与重连

异常	常见原因	处理提示
`InvalidArgumentException`	名称、消息体、超时或 TLS 配置不合法	修正参数，通常在发起网络请求前即可发现
`LogicException`	跨协程共享、未订阅就接收、重复确认或使用旧消息对象	检查客户端及消息的生命周期
`EasySwoole\Nsq\Exception\ConnectionException`	连接失败、超时、EOF、TLS 握手失败或协商被拒绝	当前连接已不可用；根据业务策略决定是否重试
`EasySwoole\Nsq\Exception\ReadException`	畸形/过大帧、未知帧类型或响应顺序错误	连接会关闭，应检查服务端与协议状态
`EasySwoole\Nsq\Exception\ServerException`	服务端返回 `E_*` 错误	错误消息保留服务端文本，连接会关闭
`EasySwoole\Nsq\Exception\LookupException`	所有 lookupd 查询均失败或响应无效	检查地址、HTTP 状态、证书和服务端响应

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

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

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

API 与源码导读

方法	用途
`connect()` / `isConnected()` / `close()`	建连、本地连接状态和关闭
`publish($topic, $body)`	普通发布
`multiPublish($topic, $messages)`	批量发布，数组中每项为非空字符串
`deferredPublish($topic, $body, $delayMilliseconds)`	延迟发布
`subscribe($topic, $channel)`	订阅 channel，初始 RDY 为 0
`consume($handler, $requeueDelayMilliseconds)`	串行消费并自动确认
`ready($count)` / `receive()`	手动设置投递额度并接收消息
`finish($message)` / `requeue($message, $delayMilliseconds)`	确认成功或重新入队
`touch($message)` / `stop()`	延长处理时限或请求退出消费循环

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

文件	阅读重点
src/Nsq.php	发布、消费循环、消息确认、协程归属和断线后状态清理
src/Config.php、src/TlsConfig.php	连接参数、TLS 校验策略和配置映射
src/Connection/Connection.php	IDENTIFY → TLS → AUTH 握手、帧长度和心跳处理
src/Connection/SocketTransport.php	协程 Socket、完整读写、TLS 升级及超时
src/Wire/Packet.php、src/Wire/Frame.php	大端编码、MPUB 长度、帧类型和纳秒时间戳
src/Lookup/Nsqlookupd.php	多节点查询、HTTPS 配置、结果校验和去重
tests/Support、tests/fixtures/tls	独立模拟服务和真实 Socket / TLS 测试夹具

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

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

easyswoole / nsq

Package info

Security

Statistics