使用Go快速验证RocketMQ是否正常的完整过程
作者：XMYX-0
最近需要验证一套 RocketMQ 环境是否可用,最开始的诉求很简单：我有一个 RocketMQ NameServer 地址,端口是默认端口 9876,想确认它到底是不是正常,所以本文给大家介绍了如何使用Go快速验证RocketMQ是否正常,需要的朋友可以参考下
前言
最近需要验证一套 RocketMQ 环境是否可用。最开始的诉求很简单：我有一个 RocketMQ NameServer 地址，端口是默认端口 9876，想确认它到底是不是正常。
很多时候我们会先用 telnet、nc 或者简单的 TCP 脚本测一下端口。这个动作当然有价值，但它只能说明：
当前机器到 RocketMQ NameServer 的网络是通的。
它不能证明 broker 正常、topic 可用，也不能证明消息真的可以成功生产和消费。
所以这次我做了一个小工具，分三层验证 RocketMQ：
- TCP 连通性测试
- 发送一条消息
- 发送并消费同一条消息，完成生产消费闭环
下面记录一下完整过程。
环境信息
本文中的内网地址做了脱敏处理，实际使用时替换成自己的 RocketMQ 地址即可。
NameServer: 10.x.x.x:9876 Go: go1.25.0 linux/amd64 RocketMQ Go Client: github.com/apache/rocketmq-client-go/v2
测试机器上 Go 环境正常：
go version
输出类似：
go version go1.25.0 linux/amd64
为什么没有继续用 Python
一开始我尝试过 Python 版本，先做 TCP 测试，再用 rocketmq-client-python 做生产消费测试。
TCP 测试是正常的：
[OK] TCP connected to 10.x.x.x:9876
但是 Python 客户端遇到了两个问题。
在 Linux 上：rocketmq-client-python 依赖 native 动态库，如果机器上缺少 librocketmq.so，会报：
rocketmq dynamic library not found
Windows 上则更直接：
rocketmq-python does not support Windows
所以最后我换成了 Go。Go 版本更适合这种轻量健康检查脚本，不需要额外处理 Python native 动态库的问题。
项目结构
新建一个目录：
mkdir rocketmq-go-healthcheck cd rocketmq-go-healthcheck
目录结构如下：
rocketmq-go-healthcheck/ ├── go.mod ├── go.sum └── main.go
go.mod
module rocketmq-go-healthcheck go 1.20 require github.com/apache/rocketmq-client-go/v2 v2.1.2 require ( github.com/emirpasic/gods v1.12.0 // indirect github.com/golang/mock v1.3.1 // indirect github.com/google/uuid v1.3.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/konsorten/go-windows-terminal-sequences v1.0.1 // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/pkg/errors v0.8.1 // indirect github.com/sirupsen/logrus v1.4.0 // indirect github.com/tidwall/gjson v1.13.0 // indirect github.com/tidwall/match v1.1.1 // indirect github.com/tidwall/pretty v1.2.0 // indirect go.uber.org/atomic v1.5.1 // indirect golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect golang.org/x/lint v0.0.0-20190930215403-16217165b5de // indirect golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f // indirect golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e // indirect gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect stathat.com/c/consistent v1.0.0 // indirect )
然后执行：
go mod tidy # 只需要go.mod 和 main.go 然后执行这个会自动生成go.sum的
核心代码
下面是完整的 main.go。
package main
import (
"context"
"flag"
"fmt"
"net"
"os"
"strings"
"sync"
"time"
rocketmq "github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/consumer"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/apache/rocketmq-client-go/v2/producer"
)
func main() {
var (
// NameServer 地址，默认端口是 9876。
namesrv = flag.String("namesrv", "10.x.x.x:9876", "RocketMQ NameServer address")
// 用于测试的 topic，需要替换成实际环境中存在的 topic。
topic = flag.String("topic", "TopicTest", "topic used for send/roundtrip")
// 测试消息使用的 tag，消费者也会按这个 tag 订阅。
tag = flag.String("tag", "healthcheck", "message tag")
// 支持三种模式：tcp 只测端口，send 只发消息，roundtrip 发送并消费。
mode = flag.String("mode", "roundtrip", "tcp, send, or roundtrip")
producerGroup = flag.String("producer-group", "PID_HEALTHCHECK_GO", "producer group")
// consumer group 默认带时间戳，避免多次测试时和已有消费者互相影响。
consumerGroup = flag.String("consumer-group", fmt.Sprintf("CID_HEALTHCHECK_GO_%d", time.Now().UnixNano()), "consumer group")
tcpTimeout = flag.Duration("tcp-timeout", 3*time.Second, "tcp connection timeout")
waitTimeout = flag.Duration("wait", 20*time.Second, "roundtrip consume wait timeout")
// consumer 启动后稍等一下再发送消息，降低订阅尚未完成导致误判的概率。
warmup = flag.Duration("warmup", 2*time.Second, "consumer warmup time before send")
)
flag.Parse()
fmt.Printf("[INFO] NameServer: %s\n", *namesrv)
fmt.Printf("[INFO] Mode: %s\n", *mode)
// 第一层检查：先测 TCP。
// 如果这里不通，优先排查网络、防火墙、安全组、端口等问题。
if err := checkTCP(*namesrv, *tcpTimeout); err != nil {
fmt.Printf("[FAIL] TCP connection failed: %v\n", err)
os.Exit(2)
}
fmt.Println("[OK] TCP connected")
switch *mode {
case "tcp":
return
case "send":
// 第二层检查：只发送一条消息，用于验证 producer 到 broker 的链路。
if err := sendMessage(*namesrv, *topic, *tag, *producerGroup, marker()); err != nil {
fmt.Printf("[FAIL] Send failed: %v\n", err)
os.Exit(3)
}
case "roundtrip":
// 第三层检查：发送并消费同一条消息，更接近真实可用性验证。
if err := roundtrip(*namesrv, *topic, *tag, *producerGroup, *consumerGroup, *waitTimeout, *warmup); err != nil {
fmt.Printf("[FAIL] Roundtrip failed: %v\n", err)
os.Exit(4)
}
default:
fmt.Printf("[FAIL] Unknown mode %q, use tcp, send, or roundtrip\n", *mode)
os.Exit(1)
}
}
func checkTCP(address string, timeout time.Duration) error {
if !strings.Contains(address, ":") {
address += ":9876"
}
start := time.Now()
conn, err := net.DialTimeout("tcp", address, timeout)
if err != nil {
return err
}
_ = conn.Close()
fmt.Printf("[OK] TCP connected to %s in %d ms\n", address, time.Since(start).Milliseconds())
return nil
}
func sendMessage(namesrv, topic, tag, group, key string) error {
p, err := rocketmq.NewProducer(
producer.WithNameServer([]string{namesrv}),
producer.WithGroupName(group),
)
if err != nil {
return err
}
if err := p.Start(); err != nil {
return err
}
defer p.Shutdown()
// 每条测试消息都带唯一 key，方便 roundtrip 模式精确识别本次发送的消息。
body := fmt.Sprintf("rocketmq-go-healthcheck %s", key)
msg := primitive.NewMessage(topic, []byte(body))
msg.WithTag(tag)
msg.WithKeys([]string{key})
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
result, err := p.SendSync(ctx, msg)
if err != nil {
return err
}
fmt.Printf("[OK] Sent message: status=%v msg_id=%s queue=%s offset=%d\n",
result.Status, result.MsgID, result.MessageQueue.String(), result.QueueOffset)
return nil
}
func roundtrip(namesrv, topic, tag, producerGroup, consumerGroup string, waitTimeout, warmup time.Duration) error {
// 本次健康检查的唯一标识，避免误消费 topic 中已有的历史消息。
key := marker()
found := make(chan *primitive.MessageExt, 1)
var once sync.Once
c, err := rocketmq.NewPushConsumer(
consumer.WithNameServer([]string{namesrv}),
consumer.WithGroupName(consumerGroup),
consumer.WithConsumerModel(consumer.Clustering),
)
if err != nil {
return err
}
// 只订阅指定 tag 的消息，减少无关消息干扰。
err = c.Subscribe(topic, consumer.MessageSelector{
Type: consumer.TAG,
Expression: tag,
}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
for _, msg := range msgs {
body := string(msg.Body)
if body == "" {
continue
}
// 只认本次发送的测试消息。
if containsKey(msg, key) || strings.Contains(body, key) {
once.Do(func() {
found <- msg
})
}
}
return consumer.ConsumeSuccess, nil
})
if err != nil {
return err
}
if err := c.Start(); err != nil {
return err
}
defer c.Shutdown()
fmt.Printf("[INFO] Consumer started: group=%s topic=%s tag=%s\n", consumerGroup, topic, tag)
time.Sleep(warmup)
// consumer 启动并完成短暂预热后，再发送测试消息。
if err := sendMessage(namesrv, topic, tag, producerGroup, key); err != nil {
return err
}
select {
case msg := <-found:
fmt.Printf("[OK] Consumed healthcheck message: msg_id=%s queue=%s offset=%d body=%s\n",
msg.MsgId, msg.Queue.String(), msg.QueueOffset, string(msg.Body))
return nil
case <-time.After(waitTimeout):
return fmt.Errorf("message was sent but not consumed within %s", waitTimeout)
}
}
func marker() string {
return fmt.Sprintf("healthcheck-%d", time.Now().UnixNano())
}
func containsKey(msg *primitive.MessageExt, key string) bool {
keys := msg.GetKeys()
return keys == key || strings.Contains(keys, key)
}
支持的测试模式
这个工具支持三种模式。
只测试 TCP 连通性
go run . -mode tcp -namesrv 10.x.x.x:9876
如果成功，会看到类似输出：
[INFO] NameServer: 10.x.x.x:9876 [INFO] Mode: tcp [OK] TCP connected to 10.x.x.x:9876 in 40 ms [OK] TCP connected
这个结果只能说明网络和端口没问题，不能说明 RocketMQ 生产消费一定正常。
只发送一条消息
go run . -mode send -namesrv 10.x.x.x:9876 -topic TopicTest
成功输出类似：
[OK] Sent message: status=SendOK msg_id=xxx queue=xxx offset=123
如果这一步失败，常见原因包括：
- topic 不存在
- broker 没有正常注册到 NameServer
- 客户端到 broker 网络不通
- ACL 权限不足
发送并消费闭环测试
go run . -namesrv 10.x.x.x:9876 -topic TopicTest
默认模式就是 roundtrip，等价于：
go run . -mode roundtrip -namesrv 10.x.x.x:9876 -topic TopicTest
成功输出类似：
[INFO] Consumer started: group=CID_HEALTHCHECK_GO_xxx topic=TopicTest tag=healthcheck [OK] Sent message: status=SendOK msg_id=xxx queue=xxx offset=123 [OK] Consumed healthcheck message: msg_id=xxx queue=xxx offset=123 body=rocketmq-go-healthcheck healthcheck-xxx
看到这类输出，基本就可以说明：
- NameServer 可连接
- broker 路由可获取
- topic 可用
- producer 可以发送消息
- consumer 可以订阅并消费消息
这比单纯测端口更有说服力。
几个实现细节
为什么先做 TCP 检查
代码一开始先调用 checkTCP。
这样做的好处是可以快速区分问题类型：
TCP 不通：优先查网络、防火墙、安全组、端口 TCP 通但发送失败：优先查 RocketMQ 路由、topic、broker、ACL 发送成功但消费失败：优先查 consumer group、tag、订阅、broker 到客户端网络
排障时分层非常重要，不然很容易一上来就陷入客户端报错细节里。
为什么 roundtrip 里先启动 consumer
代码里是先启动 consumer，再发送消息：
if err := c.Start(); err != nil {
return err
}
time.Sleep(warmup)
if err := sendMessage(namesrv, topic, tag, producerGroup, key); err != nil {
return err
}
这是为了避免消费者还没完成订阅初始化，消息就已经发出去了。虽然 RocketMQ 本身是消息队列，但对于这种一次性的健康检查脚本来说，先启动 consumer，等一小会儿，再发送消息，结果更稳定。
为什么每次消息都带唯一 marker
每次运行会生成一个类似这样的 key：
healthcheck-171xxxxxxxxxxxxx
消费回调里只接受包含这个 marker 的消息：
if containsKey(msg, key) || strings.Contains(body, key) {
found <- msg
}
这样可以避免误把 topic 里其他历史消息当成本次健康检查结果。
常见问题
TCP 通，但 send 失败
这种情况说明 NameServer 端口能访问，但 RocketMQ 链路还不完整。
可以重点检查：
topic 是否存在 broker 是否启动 broker 是否注册到了 NameServer 客户端机器是否能访问 broker 地址 是否开启了 ACL
尤其要注意：客户端连接的是 NameServer，但真正发送消息时，还会根据路由连接 broker。所以 NameServer 通，不代表 broker 一定通。
send 成功，但 roundtrip 消费不到
可以检查：
topic 是否正确 tag 是否匹配 consumer group 是否被其他程序占用 等待时间是否太短 broker 到客户端网络是否正常
可以适当把等待时间调大：
go run . -topic TopicTest -wait 60s
topic 不是 TopicTest 怎么办
直接用 -topic 参数指定真实 topic：
go run . -namesrv 10.x.x.x:9876 -topic YourRealTopic
小结
这次验证 RocketMQ 的过程可以总结成一句话：
TCP 通只能说明端口可达，生产消费闭环成功，才能更接近真实可用。
最终我用 Go 写了一个小工具，支持：
tcp：只测 NameServer 端口send：发送一条测试消息roundtrip：发送并消费同一条测试消息
实际排障时建议按这个顺序来：
go run . -mode tcp -namesrv 10.x.x.x:9876 go run . -mode send -namesrv 10.x.x.x:9876 -topic TopicTest go run . -mode roundtrip -namesrv 10.x.x.x:9876 -topic TopicTest
这样能快速判断问题到底在网络层、RocketMQ 路由层，还是生产消费链路上。
以上就是使用Go快速验证RocketMQ是否正常的完整过程的详细内容，更多关于Go验证RocketMQ是否正常的资料请关注脚本之家其它相关文章！