NSQ 1.3.0 常用 CLI 命令与工具详解

目标：系统梳理 NSQ 1.3.0 官方套件的 9 个命令行工具与 3 套 HTTP API、1 套 TCP 二进制协议，全部命令与回显均在 Ubuntu 22.04 + NSQ 1.3.0 环境实测。单机安装见《Ubuntu 22.04 安装 NSQ 1.3.0 指南》，集群形态见《NSQ 1.3.0 集群部署指南》。

一、环境信息

1.1 官方工具总览（v1.3.0 发布包 bin/ 实测清单）

⚠️ 旧教程里常见的 nsq_pubsub（HTTP 长轮询订阅器）与 nsq_writer（多节点 failover 生产器）已不在 1.3.0 发布包与源码中：官方 ChangeLog 明确 #1056 - Removed the nsq_pubsub utility（v1.3.0 源码 ChangeLog.md 实查），go install .../apps/nsq_pubsub@v1.3.0 也会报 module found, but does not contain package（实测）。

二、前置条件

nsqd --version
# nsqd v1.3.0 (built w/go1.21.5)

命令来自 /usr/local/nsq/bin（安装配置见安装篇 4.3）。启动一个干净的实验 nsqd：

nsqd

实测回显（节选）：

[nsqd] INFO: nsqd v1.3.0 (built w/go1.21.5)
[nsqd] INFO: TCP: listening on [::]:4150
[nsqd] INFO: HTTP: listening on [::]:4151

三、守护进程命令详解

3.1 nsqd：数据节点

启动示例（单机持久化）：

nsqd -data-path=/var/lib/nsq

核心参数速查（nsqd --help 实测节选）：

3.2 nsqlookupd：发现节点

nsqlookupd -broadcast-address=nsq1

3.3 nsqadmin：管理台

nsqadmin -lookupd-http-address=nsq1:4161

四、客户端工具详解（全部实测）

4.1 to_nsq：把 stdin 变成消息流

printf 'msg-from-to_nsq-1\nmsg-from-to_nsq-2\n' | to_nsq -topic=demo -nsqd-tcp-address=127.0.0.1:4150 && echo to_nsq-pub-ok
# to_nsq-pub-ok

⚠️ to_nsq 不支持 -version（实测报 flag provided but not defined: -version）。

4.2 nsq_tail：命令行消费

timeout 4 nsq_tail --topic=demo --channel=cli --nsqd-tcp-address=127.0.0.1:4150

实测回显（节选）：

2026/10/19 17:23:04 INF 1 [demo/cli] (nsq-inst:4150) connecting to nsqd
2026/10/19 17:23:04 msg-from-to_nsq-1
2026/10/19 17:23:04 msg-from-to_nsq-2

集群模式（经 lookupd，自动发现所有数据节点）：

nsq_tail --topic=orders --channel=pay --lookupd-http-address=nsq1:4161

实测回显（节选）：

INF 1 [orders/pay] (nsq1:4161) querying nsqlookupd (http://nsq1:4161/lookup?topic=orders)
INF 1 [orders/pay] (nsq3:4150) connecting to nsqd
INF 1 [orders/pay] (nsq2:4150) connecting to nsqd
2026/10/19 17:26:23 orders-A2
2026/10/19 17:26:23 orders-B1

常用参数：--topic/--channel/--nsqd-tcp-address/--lookupd-http-address（源二选一）、--max-in-flight、--print-timestamps。

4.3 nsq_to_file：订阅落盘（日志归档）

nsq_to_file --topic=demo --channel=filech --output-dir=/tmp/nsqfiles --nsqd-tcp-address=127.0.0.1:4150

发布两条后观察输出（实测）：

ls -l /tmp/nsqfiles/
# -rw-r--r-- 1 root root 22 Oct 19 17:24 /tmp/nsqfiles/demo.nsq-inst.2026-10-19_17.log
find /tmp/nsqfiles -type f -exec cat {} \;
# file-msg-1
# file-msg-2

文件名格式为 <topic>.<主机名>.<日期>_<小时>.log，按小时滚动。常用参数：--gzip、--rotate-size、--rotate-interval、--skip-empty-files、--select-by-tag。

4.4 nsq_to_http：订阅转 HTTP 回调

先起一个接收端（实验用 Python 单脚本模拟业务回调），再启动转发器：

nohup timeout 6 nsq_to_http --topic=demo --channel=httpch2 --nsqd-tcp-address=127.0.0.1:4150 -post=http://127.0.0.1:8105/ingest >/tmp/n2h2.log 2>&1 &
curl -s -d http-msg-C 'http://127.0.0.1:4151/pub?topic=demo' >/dev/null

接收端实测回显：

RECV: /ingest http-msg-C

⚠️ 排障实录：转发目标参数是 -post/-get（可多次），不是 --http-post-address——用错参数名会直接打印 usage 退出（实测）。--status-every N 可控制进度日志频率。

4.5 nsq_to_nsq：跨实例桥接复制

从源（直连 nsqd 或经 lookupd）订阅，发布到目标 nsqd。直连源模式：

nsq_to_nsq -nsqd-tcp-address=127.0.0.1:4150 -topic=demo -destination-nsqd-tcp-address=127.0.0.1:14150

lookupd 源模式（⚠️ 必须先让源 topic 在 lookupd 有注册——先发一条消息再启动，否则消费端要等默认 60s 轮询才能发现，实测曾因此空跑 8 秒零转发）：

curl -s -d seed-topic 'http://127.0.0.1:14153/pub?topic=demo' >/dev/null
nsq_to_nsq --lookupd-http-address=127.0.0.1:4161 --topic=demo --destination-nsqd-tcp-address=127.0.0.1:14150

转发完成后核对目标实例统计：

curl -s 'http://127.0.0.1:14151/stats?format=json'
# {"topics":[{"topic_name":"demo","channels":[],"depth":0,"memory_depth":0,"in_flight_count":0,"message_count":3,...}]}

其他实用参数：-destination-topic（改写目标 topic）、-require-json-field/-require-json-value/-whitelist-json-field（JSON 过滤）、-mode（多目标发布策略：round-robin/hostpool/epsilon-greedy）。

4.6 nsq_stat：聚合统计

nsq_stat -interval 1s -count 1 --lookupd-http-address=nsq1:4161 --topic=orders --channel=pay

实测回显：

Connections per channel, per nsqd node

depth    mem-depth  inflt    def    re-q    timeout    msgs    e2e
0        0          0        0      0       0          6       0.0ms
nsq3:4150 [client: 198.18.x.x]
total 1 connections

⚠️ nsq_stat 不带 -count 时会持续刷新不退出（实测进入循环输出）；脚本化使用务必加 -interval 1s -count 1。

五、HTTP API 实测

5.1 nsqd HTTP API（默认 4151）

v1.3.0 路由全表（源码 nsqd/http.go 实查）：

实测（创建→发布→批量→统计→清空→暂停→删除）：

curl -s -o /dev/null -w '%{http_code}\n' -X POST 'http://127.0.0.1:4151/topic/create?topic=api_demo'
# 200
curl -s -d 'api-msg-1' 'http://127.0.0.1:4151/pub?topic=api_demo'
# OK
printf 'mpub-a\nmpub-b\nmpub-c\n' | curl -s --data-binary @- 'http://127.0.0.1:4151/mpub?topic=api_demo'
# OK
curl -s 'http://127.0.0.1:4151/stats?topic=api_demo'

实测回显（节选）：

nsqd v1.3.0 (built w/go1.21.5)
start_time 2026-10-19T17:47:44Z
uptime 4m16.729938s

Topology
------------------------------------------------------------
Topic/Channel                Info
------------------------------------------------------------
api_demo                     depth: 4    be-depth: 0    msgs: 4     e2e%:
  * ch1                      depth: 4    be-depth: 0    inflt: 0    def: 0    re-q: 0    timeout: 0    msgs: 4     e2e%: ***

curl -s -X POST 'http://127.0.0.1:4151/topic/empty?topic=api_demo'
# （HTTP 200，空响应体）
curl -s -X POST 'http://127.0.0.1:4151/topic/pause?topic=api_demo'
# （HTTP 200）
curl -s 'http://127.0.0.1:4151/stats?format=json&topicsPattern=api_demo' | python3 -c 'import json,sys; t=json.load(sys.stdin)["topics"][0]; print("paused:",t["paused"])'
# paused: True
curl -s http://127.0.0.1:4151/config/max_msg_timeout
# 900000000000
curl -s http://127.0.0.1:4151/config/mem_queue_size
# 10000
curl -s http://127.0.0.1:4151/info
# {"version":"1.3.0","broadcast_address":"nsq-inst","hostname":"nsq-inst","http_port":4151,"tcp_port":4150,"start_time":1792432064}
curl -s -o /dev/null -w '%{http_code}\n' -X POST 'http://127.0.0.1:4151/topic/delete?topic=api_demo'
# 200

⚠️ 两个实测排障点：

1. /topic/create 只认 URL 查询参数——用 curl -d 'topic=api_demo'（POST body）会报 {"message":"MISSING_ARG_TOPIC"}；
2. 旧路径已死——1.0 前的 /empty、/pause 在 1.3.0 返回 {"message":"NOT_FOUND"}，必须用上表的资源化路径（很多旧博客仍写 /empty?topic=，直接照抄必踩坑）。

5.2 nsqlookupd HTTP API（默认 4161）

v1.3.0 路由全表（源码 nsqlookupd/http.go 实查）：/ping、/info、/topics、/channels?topic=、/lookup?topic=、/nodes、/topic/delete?topic=&node=、/topic/tombstone?topic=&node=。

实测：

curl -s http://127.0.0.1:4161/ping
# OK
curl -s http://127.0.0.1:4161/info
# {"version":"1.3.0","broadcast_address":"nsq-inst","hostname":"nsq-inst","http_port":4161,"tcp_port":4160,"start_time":1792432572}
curl -s http://127.0.0.1:4161/topics
# {"topics":["demo"]}
curl -s 'http://127.0.0.1:4161/channels?topic=demo'
# {"channels":[]}
curl -s 'http://127.0.0.1:4161/lookup?topic=demo'

实测回显（节选）：

{"channels":["nsq_to_nsq"],"meta":{"delivered":true},"producers":[{"remote_address":"127.0.0.1:59124","hostname":"nsq-inst","broadcast_address":"127.0.0.1","tcp_port":14152,"http_port":14153,"version":"1.3.0"}]}

curl -s http://127.0.0.1:4161/nodes
# {"producers":[{...,"broadcast_address":"nsq2","tcp_port":4150,"http_port":4151,...}]}

下线发布者（tombstone）：必须带 node 参数（实测缺省报 {"message":"MISSING_ARG_NODE"}），执行后该生产者在 -tombstone-lifetime（默认 45s）内不再被 lookupd 通告：

curl -s -X POST 'http://127.0.0.1:4161/topic/tombstone?topic=demo&node=127.0.0.1'
# （HTTP 200，空响应体）
curl -s 'http://127.0.0.1:4161/lookup?topic=demo'
# {"message":"TOPIC_NOT_FOUND"}

5.3 nsqadmin HTTP API（默认 4171）

v1.3.0 路由全表（源码 nsqadmin/http.go 实查）：/api/topics、/api/topics/:topic、/api/topics/:topic/:channel、/api/nodes、/api/nodes/:node、/api/counter?topic_name=&channel_name=、/config，以及管理动作（POST JSON 到 /api/topics/:topic：empty_topic、empty_channel、delete_topic、delete_channel、pause_channel、unpause_channel、tombstone_topic_producer）。

实测（视图层）：

curl -s http://nsq1:4171/api/nodes
# {"nodes":[{..."broadcast_address":"nsq2","tcp_port":4150,...},{..."broadcast_address":"nsq3","tcp_port":4150,...}]}
curl -s http://nsq1:4171/api/topics

实测回显（节选）：

{"topics":[{"topic_name":"orders","channels":[{"channel_name":"pay","paused":false,...}],"nodes":["nsq2","nsq3"],"depth":6,"memory_depth":6,...}]}

curl -s 'http://127.0.0.1:4171/api/topics/demo'
# {"topic":"demo","channels":[...],"nodes":["127.0.0.1"],"stats":[{...,"depth":0,"message_count":8,...}]}

⚠️ nsqadmin 没有 /api/stats 路径（实测返回 {"message":"NOT_FOUND"}）；topic/channel 统计走 /api/topics/:topic。浏览器访问 http://<nsqadmin>:4171/ 即管理台（前端渲染，接口层可按上表验证）。

六、TCP 二进制协议实测

6.1 帧格式与握手

TCP 协议（nsq.io/clients/tcp_protocol_spec 口径 + 实测验证）：

1. 客户端连接后必须先发 4 字节魔数 V2（两个空格+V2）；
2. 之后每帧均为 [4 字节大端长度][载荷]，命令帧载荷即命令文本+body（PUB/MPUB body 前 4 字节为 body 长度）；
3. 服务端响应帧为 [4 字节帧长][4 字节帧类型][载荷]，帧类型：0=Response、1=Error、2=Message。

不握手直接发命令，实测被拒：

printf 'PUB wire_demo\n\x00\x00\x00\x0Bwire hello!' | timeout 3 nc 127.0.0.1 4150 | od -A d -t x1z | head -3

实测回显：

0000000 00 00 00 12  00 00 00 01  45 5f 42 41 44 5f 50 52 4f 54 4f 43 4f 4c
         └─帧长 18──┘ └类型 1─┘ └──── E_BAD_PROTOCOL ────┘

补上魔数后重新发布（\x00\x00\x00\x0B 即 11 字节 body 长度）：

printf '  V2PUB wire_demo\n\x00\x00\x00\x0Bwire hello!' | timeout 3 nc 127.0.0.1 4150 | od -A d -t x1z | head -3

实测回显：

0000000 00 00 00 06  00 00 00 00  4f 4b
         └帧长 6──┘ └类型 0┘ └OK┘

6.2 原生 TCP 订阅（SUB/RDY/FIN）

用 Python 演示完整消费流程（魔数→SUB→RDY→读 MSG 帧→FIN）：

# /tmp/sub_demo.py
import socket, struct, urllib.request

def recvn(sock, n):
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise EOFError
        buf += chunk
    return buf

def read_frame(sock):
    size = struct.unpack(">i", recvn(sock, 4))[0]
    payload = recvn(sock, size)
    ftype = struct.unpack(">i", payload[:4])[0]
    return ftype, payload[4:]

s = socket.create_connection(("127.0.0.1", 4150), timeout=5)
s.sendall(b"  V2")
s.sendall(b"SUB wire_demo ch1\n")
s.sendall(b"RDY 1\n")
urllib.request.urlopen(urllib.request.Request(
    "http://127.0.0.1:4151/pub?topic=wire_demo", data=b"tcp-sub-hello", method="POST"))
ftype, body = read_frame(s)
print("SUB response frame: type=%d body=%r" % (ftype, body))
ftype, msg = read_frame(s)
ts = struct.unpack(">q", msg[:8])[0]
attempts = struct.unpack(">h", msg[8:10])[0]
mid = msg[10:26].hex()
mbody = msg[26:]
print("MSG frame: type=%d ts=%d attempts=%d id=%s body=%r" % (ftype, ts, attempts, mid, mbody.decode()))
s.sendall(b"FIN " + mid.encode() + b"\n")
s.close()

实测回显：

SUB response frame: type=0 body=b'OK'
MSG frame: type=2 ts=1792432493854215318 attempts=1 id=0e790cb55710e000 body='wire hello!'

MSG 帧结构实测验证：[2 字节时间戳后的 8 字节纳秒时间戳][2 字节投递次数][16 字节消息 ID][消息体]——收到的 attempts=1、id 为 16 字节 hex、body 完整还原。

💡 说明：本例消费到的是此前 6.1 经 TCP 发布的 wire hello!（channel 队列积压先到先投），体现了 NSQ「消息先进 channel 队列、消费者按 RDY 拉取推送」的语义。

七、注意事项

1. 端口分工：TCP 生产消费走 4150，HTTP/管理走 4151；curl 混用会 connection refused。
2. 参数名首横杠：官方工具 -topic（单横杠）与 GNU 风格 --topic 通用（Go flag 库等价），文档混用不属错误；但参数名拼错不会报 unknown flag 之外的额外提示，直接打印 usage 退出。
3. /topic/create 查询参数 vs body：多数管理接口只认查询参数，/pub 同样（消息体才是 body）。
4. 工具退出语义：nsq_tail/nsq_to_* 都是常驻进程，脚本里用 timeout 包裹；nsq_stat 要 -count 1。
5. nsq_to_nsq 发现延迟：lookupd 源模式先有 topic 注册再启动（见 4.5）。
6. 消息幂等：任何消费端工具/客户端都可能重复投递（至少一次语义，官方保证页口径），业务侧按消息 ID 去重。
7. -version 支持面：8 个工具支持；to_nsq 不支持。

八、常见问题（FAQ）

Q1：nsq_pubsub 在哪？ v1.3.0 已移除（ChangeLog #1056，源码与发布包实测均无该文件）。HTTP 长轮询消费的替代：用客户端库（go-nsq/pynsq/nsqjs）或 nsq_tail/nsq_to_http。

Q2：/empty?topic=x 返回 NOT_FOUND？ 1.x 新路径是 /topic/empty（channel 维度 /channel/empty）。GET 带查询参数的 /empty 在 1.3.0 路由表中不存在。

Q3：MISSING_ARG_TOPIC / MISSING_ARG_NODE？ 对应接口缺少必需查询参数：topic 管理类接口必须 ?topic=；lookupd 的 /topic/tombstone、/topic/delete 必须同时带 ?topic=&node=（实测）。

Q4：stats 里 depth 与 be-depth 是什么？ depth 是 channel 当前积压（内存+磁盘），be-depth 是磁盘队列积压；inflt 在途、def 延迟队列、re-q 重投计数（实测表头口径）。

Q5：如何看消息是否真的落盘了？ 把 -mem-queue-size 设 0 后再发布，/var/lib/nsq 会生成按 topic 命名的磁盘队列文件；/stats 中 be-depth 同步增长。

Q6：工具与 nsqd 版本必须一致吗？ 官方客户端（go-nsq）与服务端按协议协商兼容；实测 1.3.0 全套工具对 1.3.0 服务端工作正常。跨大版本混用未在本文验证范围。

九、参考链接

• TCP 协议规范：https://nsq.io/clients/tcp_protocol_spec.html
• HTTP 接口口径：https://nsq.io/clients/http_api.html
• 官方组件文档：https://nsq.io/components/nsqd.html
• v1.3.0 源码：https://github.com/nsqio/nsq/tree/v1.3.0
• 站内相关：《NSQ 1.3.0 介绍与选型指南》、《Ubuntu 22.04 安装 NSQ 1.3.0 指南》、《NSQ 1.3.0 集群部署指南》