RocketMQ在GoFrame框架中的实现方案

项目概述

在GoFrame框架中实现RocketMQ消息队列的完整解决方案。采用分层架构设计,包含配置管理、基础封装、业务逻辑和接口控制四个层次。

技术特点

配置管理

• 支持YAML配置文件管理连接信息
• 可动态切换消息队列类型(RocketMQ/RabbitMQ)
• 包含nameServer、group等核心配置项

基础封装(pkg层)

• 使用sync.Once确保单例初始化
• 封装生产者消费者实例管理
• 提供优雅关闭机制
• 统一的错误处理

业务逻辑(logic层)

• 实现消息发送与消费的核心逻辑
• 支持消息状态跟踪和持久化
• 提供消息重试机制
• 完整的日志记录

接口控制(controller层)

• 提供HTTP接口操作消息
• 统一的响应格式
• 完善的参数校验
• 友好的错误提示

应用场景

• 异步任务处理
• 系统间解耦
• 消息通知
• 延迟任务
• 流量削峰

扩展性

• 支持自定义消息处理逻辑
• 可扩展消息过滤规则
• 支持多种消息队列切换
• 便于二次开发

注意事项

• 需提前初始化连接
• 注意消息幂等性处理
• 建议添加监控告警
• 关注消息积压情况

代码实现

配置文件

# 消息队列配置
mq:
  # 消息队列类型: rocketmq 或 rabbitmq
  type: "rabbitmq"
  # 是否启用消息队列
  enabled: true

rocketmq:
  nameServer: "127.0.0.1:9876"
  producerGroup: "myProducerGroup"
  consumerGroup: "myConsumerGroup"
  brokerAddress: "127.0.0.1:10911"  # 添加 broker 地址

rabbitmq:
  url: "amqp://wanghaibin:wanghaibin@127.0.0.1:5672/"
  exchange: "gf_exchange"
  dlx_exchange: "gf_dlx_exchange"    # 新增：死信交换机
  queue: "gf_queue"
  delay_queue: "gf_delay_queue"      # 新增：延迟队列
  routingKey: "gf_key"
  vhost: "/"

基础包实现

package rocketmq

import (
    "context"
    "sync"

    "github.com/apache/rocketmq-client-go/v2"
    "github.com/apache/rocketmq-client-go/v2/consumer"
    "github.com/apache/rocketmq-client-go/v2/producer"
    "github.com/gogf/gf/v2/frame/g"
)

// 全局变量定义
var (
    once sync.Once
    mqProducer rocketmq.Producer
    mqConsumer rocketmq.PushConsumer
)

func Initialize() error {
    once.Do(func() {
        var err error
        ctx := context.Background()
        nameServer := g.Cfg().MustGet(ctx, "rocketmq.nameServer").String()
        groupName := g.Cfg().MustGet(ctx, "rocketmq.groupName").String()

        mqProducer, err = rocketmq.NewProducer(
            producer.WithNameServer([]string{nameServer}),
            producer.WithGroupName(groupName),
            producer.WithRetry(2),
        )
        if err != nil {
            g.Log().Fatalf(ctx, "Failed to create producer: %v", err)
        }

        if err := mqProducer.Start(); err != nil {
            g.Log().Fatalf(ctx, "Failed to start producer: %v", err)
        }

        mqConsumer, err = rocketmq.NewPushConsumer(
            consumer.WithNameServer([]string{nameServer}),
            consumer.WithGroupName(groupName),
        )
        if err != nil {
            g.Log().Fatalf(ctx, "Failed to create consumer: %v", err)
        }
    })
    return nil
}

func GetProducer() rocketmq.Producer {
    return mqProducer
}

func GetConsumer() rocketmq.PushConsumer {
    return mqConsumer
}

func Cleanup() {
    if mqProducer != nil {
        mqProducer.Shutdown()
    }
    if mqConsumer != nil {
        mqConsumer.Shutdown()
    }
}

业务逻辑实现

package rocketmsg

import (
    "context"
    "gf_new_web/internal/model"
    "gf_new_web/internal/pkg/rocketmq"
    "gf_new_web/internal/service"

    mqConsumer "github.com/apache/rocketmq-client-go/v2/consumer"
    "github.com/apache/rocketmq-client-go/v2/primitive"
    "github.com/gogf/gf/v2/frame/g"
    "github.com/gogf/gf/v2/os/gtime"
)

type sRocketMsg struct{}

func init() {
    service.RegisterRocketMsg(New())
}

func New() *sRocketMsg {
    return &sRocketMsg{}
}

func (s *sRocketMsg) ProduceMessage(ctx context.Context, topic, tags, content string) error {
    producer := rocketmq.GetProducer()
    msg := &primitive.Message{
        Topic: topic,
        Body:  []byte(content),
    }
    msg.WithTag(tags)

    result, err := producer.SendSync(ctx, msg)
    if err != nil {
        g.Log().Error(ctx, "发送消息失败", err)
        return err
    }

    message := &model.Message{
        Topic:      topic,
        Content:    content,
        Tags:       tags,
        Status:     "sent",
        MessageID:  result.MsgID,
        CreateTime: gtime.Now().String(),
        UpdateTime: gtime.Now().String(),
    }

    if err := message.Save(ctx); err != nil {
        g.Log().Error(ctx, "保存消息记录失败", err)
        return err
    }

    return nil
}

func (s *sRocketMsg) ConsumeMessages(ctx context.Context, topic, tags string) error {
    consumer := rocketmq.GetConsumer()

    err := consumer.Subscribe(topic, mqConsumer.MessageSelector{
        Type:       mqConsumer.TAG,
        Expression: tags,
    }, func(ctx context.Context, msgs ...*primitive.MessageExt) (mqConsumer.ConsumeResult, error) {
        for _, msg := range msgs {
            message := &model.Message{
                Topic:      msg.Topic,
                Content:    string(msg.Body),
                Tags:       msg.GetTags(),
                Status:     "received",
                CreateTime: gtime.Now().String(),
                UpdateTime: gtime.Now().String(),
            }

            if err := s.handleMessage(ctx, msg); err != nil {
                message.Status = "failed"
                message.Save(ctx)
                return mqConsumer.ConsumeRetryLater, err
            } else {
                message.Status = "processed"
                message.Save(ctx)
            }
        }
        return mqConsumer.ConsumeSuccess, nil
    })

    if err != nil {
        return err
    }

    return consumer.Start()
}

func (s *sRocketMsg) handleMessage(ctx context.Context, msg *primitive.MessageExt) error {
    g.Log().Info(ctx, "Processing message", "Message ID", msg.MsgId, "Body", string(msg.Body))
    return nil
}

服务接口定义

package service

import (
    "context"
)

type (
    IRocketMsg interface {
        ProduceMessage(ctx context.Context, topic string, tags string, content string) error
        ConsumeMessages(ctx context.Context, topic string, tags string) error
    }
)

var (
    localRocketMsg IRocketMsg
)

func RocketMsg() IRocketMsg {
    if localRocketMsg == nil {
        panic("implement not found for interface IRocketMsg, forgot register?")
    }
    return localRocketMsg
}

func RegisterRocketMsg(i IRocketMsg) {
    localRocketMsg = i
}

控制器实现

package front

import (
    "gf_new_web/internal/service"
    "github.com/gogf/gf/v2/frame/g"
    "github.com/gogf/gf/v2/net/ghttp"
)

var (
    RocketMsg = cRocketMsg{}
)

type cRocketMsg struct{}

func (c *cRocketMsg) ProduceRocketMsg(r *ghttp.Request) {
    var (
        topic   = r.Get("topic").String()
        tags    = r.Get("tags").String()
        content = r.Get("content").String()
    )
    err := service.RocketMsg().ProduceMessage(r.GetCtx(), topic, tags, content)
    if err != nil {
        g.Log().Error(r.GetCtx(), err)
        r.Response.WriteJson(g.Map{
            "code": -1,
            "msg":  err.Error(),
        })
        return
    }
    r.Response.WriteJson(g.Map{
        "code": 0,
        "msg":  "消息发送成功",
    })
}

func (c *cRocketMsg) ConsumeRocketMsg(r *ghttp.Request) {
    var (
        topic = r.Get("topic").String()
        tags  = r.Get("tags").String()
    )
    err := service.RocketMsg().ConsumeMessages(r.GetCtx(), topic, tags)
    if err != nil {
        g.Log().Error(r.GetCtx(), err)
        r.Response.WriteJson(g.Map{
            "code": -1,
            "msg":  err.Error(),
        })
        return
    }
    r.Response.WriteJson(g.Map{
        "code": 0,
        "msg":  "消息消费成功",
    })
}