使用Go快速验证RocketMQ是否正常的完整过程
更新时间：2026年06月28日 13:45:05 作者：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是否正常的资料请关注脚本之家其它相关文章！
相关文章
详解Golang time包中的time.Duration类型
在日常开发过程中，会频繁遇到对时间进行操作的场景，使用 Golang 中的 time 包可以很方便地实现对时间的相关操作，本文讲解一下 time 包中的 time.Duration 类型,需要的朋友可以参考下2023-07-07
Casbin是用于Golang项目的功能强大且高效的开源访问控制库。本文主要介绍了go语言使用Casbin实现角色的权限控制，感兴趣的可以了解下2021-06-06
go官方介绍go内存模型的时候说：探究在什么条件下，goroutine 在读取一个变量的值的时，能够看到其它 goroutine 对这个变量进行的写的结果，Go内存模型规定了一些条件，在这些条件下，在一个goroutine中读取变量返回的值能够确保是另一个goroutine中对该变量写入的值2022-12-12
这篇文章主要介绍了golang的基础语法和常用开发工具,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-12-12
Golang中Map遍历方法解析,介绍for...range循环遍历无序键值对集合map的常见方式,强调协程环境下的同步保护措施2026-06-06
这篇文章主要介绍了Golang学习笔记（二）：类型、变量、常量,本文讲解了基本类型、保留字、变量、常量、枚举、运算符、指针、分组声明等内容,需要的朋友可以参考下2015-05-05
本文主要介绍了Go语言单例模式详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2023-03-03
如果大家系统的学过C++、Java等语言以及面向对象的话，相信应该对多态不会陌生。多态是面向对象范畴当中经常使用并且非常好用的一个功能，它主要是用在强类型语言当中，像是Python这样的弱类型语言，变量的类型可以随意变化，也没有任何限制，其实区别不是很大2021-06-06
本文主要介绍了Go语言nil标识符(空值/零值)，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2023-03-03
如何在 ubuntu linux 上配置 go 语言的 qt 开发环境
这篇文章主要介绍了如何在 ubuntu linux 上配置 go 语言的 qt 开发环境,本文分步骤通过实例代码相结合给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-04-04
最新评论