beego无ioc容器和自动装配机制，需手动初始化rocketmq consumer：调用newpushconsumer创建实例，subscribe注册监听逻辑，再显式start启动；若顺序错误、未设groupname或忽略acl配置，将导致消费失败。
Beego 框架本身不提供 RocketMQ 官方集成支持，也没有 beego-rocketmq 这类标准 starter 包。想在 Beego 项目中用 RocketMQ，得直接对接原生 Go 客户端 github.com/apache/rocketmq-client-go/v2，并自行管理连接生命周期、错误重试、消费者组注册等细节。
为什么不能像 Spring Boot 那样自动装配？
Spring Boot 的 rocketmq-spring-boot-starter 封装了自动配置、@RocketMQMessageListener 注解扫描、上下文生命周期绑定等功能；而 Beego 是轻量级 MVC 框架，没有类似 Spring 的 IoC 容器和 AOP 基础，所有消息组件必须手动初始化、启动、关闭。
常见误操作包括：
- 在
main()中直接 new Consumer 后未调用Start()，导致监听静默失败 - 把
Consumer实例声明为局部变量，函数返回后被 GC，实际未运行 - 未设置
WithGroupName()，使用默认组名导致多个实例重复消费
如何正确初始化 RocketMQ Consumer（Go 客户端 v2）
推荐在 Beego 的 app.go 或独立的 mq/init.go 中完成初始化，并挂载到全局变量或 Beego App 的 Config 中：
import (
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/consumer"
"github.com/apache/rocketmq-client-go/v2/primitive"
)
var MqConsumer rocketmq.Consumer
func InitRocketMQ() error {
c, err := rocketmq.NewPushConsumer(
consumer.WithGroupName("beego-consumer-group"),
consumer.WithNameServer([]string{"127.0.0.1:9876"}),
consumer.WithCredentials(primitive.Credentials{
AccessKey: "",
SecretKey: "",
}),
)
if err != nil {
return err
}
// 必须显式注册消息处理逻辑
err = c.Subscribe("test-topic", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
for _, msg := range msgs {
log.Printf("received: %s", string(msg.Body))
}
return consumer.ConsumeSuccess, nil
})
if err != nil {
return err
}
// 必须调用 Start 才真正开始拉取消息
err = c.Start()
if err != nil {
return err
}
MqConsumer = c
return nil
}
注意：
Apache Superset Dashboard and SQL Exploration Skill
下载
Apache Superset 是一个广泛采用的开源 BI 平台，用于 SQL 探索、图表构建和仪表板交付。当代理需要查询仓库数据、组装仪表板或使用成熟的分析界面解释指标而不是临时笔记本代码时，此技能非常有用。
-
Subscribe()必须在Start()之前调用，否则订阅无效 - 若启用 ACL，
AccessKey/SecretKey必须非空；否则可省略WithCredentials - Topic 名称需提前在 RocketMQ 控制台或 CLI 创建，客户端不会自动建 Topic
如何在 Beego Controller 中安全发送消息
不要每次请求都 new Producer —— 连接开销大、易触发限流。应复用单例 Producer：
var MqProducer rocketmq.Producer
func InitRocketMQProducer() error {
p, err := rocketmq.NewProducer(
producer.WithGroupName("beego-producer-group"),
producer.WithNameServer([]string{"127.0.0.1:9876"}),
)
if err != nil {
return err
}
err = p.Start()
if err != nil {
return err
}
MqProducer = p
return nil
}
// 在 Controller 方法中调用
func (this *MainController) SendMsg() {
msg := primitive.NewMessage("test-topic", []byte("hello from beego"))
result, err := MqProducer.SendSync(context.Background(), msg)
if err != nil {
this.Data["json"] = map[string]interface{}{"err": err.Error()}
this.ServeJSON()
return
}
this.Data["json"] = map[string]interface{}{"status": "ok", "msgId": result.MsgID}
this.ServeJSON()
}
关键点：
-
SendSync是阻塞调用，适合对延迟敏感且能容忍失败重试的场景；高吞吐建议用SendAsync+ 回调 - 务必检查
result.Status == primitive.SendOK，仅靠err == nil不足以判断发送成功（比如 broker 返回SEND_TIMEOUT时 err 为 nil） - Producer 必须在应用退出前调用
Shutdown()，否则可能丢消息或连接泄漏
Beego 应用退出时如何优雅关闭 RocketMQ 客户端
Beego 没有内置的 shutdown hook，需在 main() 中监听 OS 信号，并主动关闭：
func main() {
if err := mq.InitRocketMQ(); err != nil {
log.Fatal(err)
}
if err := mq.InitRocketMQProducer(); err != nil {
log.Fatal(err)
}
// 启动 Beego
beego.Run()
// 此处不会执行 —— beego.Run() 是阻塞的
// 正确做法：用 goroutine + channel 监听 os.Interrupt
}
更稳妥的做法是改用 beego.BeeApp.Run() 并自己控制主循环，或在 app.go 的 init() 中注册 os.Interrupt 信号处理器：
func init() {
signal.Notify(signalChan, os.Interrupt, os.Kill)
go func() {
<p>最容易被忽略的是：Broker 默认只保留 3 天消息，且 Consumer offset 提交是异步的。如果没调用 <code>Shutdown()</code>，最后一次 offset 可能未持久化，重启后会重复消费或跳过部分消息。</p>
