RabbitMQ消息队列封装
在目录internal/pkg/rabbitmq/rabbitmq.go
# 消息队列配置
mq:# 消息队列类型: rocketmq 或 rabbitmqtype: "rabbitmq"# 是否启用消息队列enabled: truerocketmq: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 rabbitmqimport ("context""fmt""time""github.com/gogf/gf/v2/frame/g"amqp "github.com/rabbitmq/amqp091-go"
)var (// conn RabbitMQ连接实例conn *amqp.Connection// channel RabbitMQ通道实例channel *amqp.Channel
)// Initialize 初始化 RabbitMQ 连接和通道
// 包括:建立连接、创建通道、声明交换机和队列、建立绑定关系
func Initialize() {var err errorctx := context.Background()// 从配置文件获取RabbitMQ连接URLurl := g.Cfg().MustGet(ctx, "rabbitmq.url").String()// 建立RabbitMQ连接conn, err = amqp.Dial(url)if err != nil {g.Log().Fatalf(ctx, "Failed to connect to RabbitMQ: %v", err)}// 创建通道channel, err = conn.Channel()if err != nil {g.Log().Fatalf(ctx, "Failed to open channel: %v", err)}// 1. 声明主交换机// 类型:direct,持久化:true,自动删除:false,内部的:false,非阻塞:falseerr = channel.ExchangeDeclare(g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(),"direct", // 交换机类型true, // 持久化false, // 自动删除false, // 内部的false, // 非阻塞nil, // 参数)if err != nil {g.Log().Fatalf(ctx, "Failed to declare main exchange: %v", err)}// 2. 声明死信交换机(DLX)// 用于处理无法被正常消费的消息err = channel.ExchangeDeclare(g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),"direct",true,false,false,false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to declare DLX exchange: %v", err)}// 3. 声明主队列// 持久化:true,非自动删除,非排他,非阻塞_, err = channel.QueueDeclare(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),true, // 持久化false, // 自动删除false, // 排他的false, // 非阻塞nil, // 参数)if err != nil {g.Log().Fatalf(ctx, "Failed to declare main queue: %v", err)}// 4. 声明延迟队列// 配置死信交换机参数args := amqp.Table{"x-dead-letter-exchange": g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),"x-dead-letter-routing-key": g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),}_, err = channel.QueueDeclare(g.Cfg().MustGet(ctx, "rabbitmq.delay_queue").String(),true,false,false,false,args,)if err != nil {g.Log().Fatalf(ctx, "Failed to declare delay queue: %v", err)}// 5. 绑定主队列到主交换机err = channel.QueueBind(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(), // 队列名g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(), // 路由键g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(), // 交换机名false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to bind main queue: %v", err)}// 6. 绑定主队列到死信交换机err = channel.QueueBind(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),g.Cfg().MustGet(ctx, "rabbitmq.dlx_exchange").String(),false,nil,)if err != nil {g.Log().Fatalf(ctx, "Failed to bind queue to DLX: %v", err)}g.Log().Info(ctx, "RabbitMQ initialized successfully")
}// PublishMessage 发布消息到RabbitMQ
// 参数:
// - ctx: 上下文
// - message: 要发送的消息内容
// 返回:
// - error: 发送错误,如果成功则为nil
func PublishMessage(ctx context.Context, message string) error {// 创建带超时的上下文ctxTimeout, cancel := context.WithTimeout(ctx, 5*time.Second)defer cancel()// 发布消息到指定的交换机和路由err := channel.PublishWithContext(ctxTimeout,g.Cfg().MustGet(ctx, "rabbitmq.exchange").String(),g.Cfg().MustGet(ctx, "rabbitmq.routingKey").String(),false, // mandatoryfalse, // immediateamqp.Publishing{ContentType: "text/plain",Body: []byte(message),},)if err != nil {return fmt.Errorf("failed to publish message: %v", err)}return nil
}// ConsumeMessages 消费队列中的消息
// 参数:
// - ctx: 上下文
// - handler: 消息处理函数
// 返回:
// - error: 消费错误,如果成功则为nil
func ConsumeMessages(ctx context.Context, handler func(string) error) error {messages, err := channel.Consume(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),"", // consumerfalse, // auto-ackfalse, // exclusivefalse, // no-localfalse, // no-waitnil, // args)if err != nil {return fmt.Errorf("failed to register a consumer: %v", err)}// 启动goroutine处理消息go func() {for msg := range messages {err := handler(string(msg.Body))if err != nil {g.Log().Errorf(ctx, "Error handling message: %v", err)msg.Nack(false, true) // 处理失败,消息重新入队} else {msg.Ack(false) // 处理成功,确认消息}}}()return nil
}// Cleanup 清理RabbitMQ连接和通道
func Cleanup() {if channel != nil {channel.Close()}if conn != nil {conn.Close()}
}// GetChannel 获取RabbitMQ通道实例
func GetChannel() *amqp.Channel {return channel
}// PurgeQueue 清空指定队列中的所有消息
// 参数:
// - ctx: 上下文
// 返回:
// - error: 清空错误,如果成功则为nil
func PurgeQueue(ctx context.Context) error {_, err := channel.QueuePurge(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),false, // no-wait)return err
}// PublishDelayMessage 发送延迟消息
// 参数:
// - ctx: 上下文
// - message: 消息内容
// - delaySeconds: 延迟秒数
// 返回:
// - error: 发送错误,如果成功则为nil
func PublishDelayMessage(ctx context.Context, message string, delaySeconds int) error {return channel.PublishWithContext(ctx,"", // 默认交换机g.Cfg().MustGet(ctx, "rabbitmq.delay_queue").String(), // 延迟队列false,false,amqp.Publishing{ContentType: "text/plain",Body: []byte(message),Expiration: fmt.Sprintf("%d", delaySeconds*1000), // 转换为毫秒},)
}// GetQueueLength 获取队列中的消息数量
// 参数:
// - ctx: 上下文
// 返回:
// - int: 消息数量
// - error: 获取错误,如果成功则为nil
func GetQueueLength(ctx context.Context) (int, error) {queue, err := channel.QueueInspect(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),)if err != nil {return 0, fmt.Errorf("failed to inspect queue: %v", err)}return queue.Messages, nil
}
logic逻辑的实现
package rabbitmqmsgimport ("context""fmt""gf_new_web/internal/pkg/rabbitmq""gf_new_web/internal/service""github.com/gogf/gf/v2/frame/g"
)// sRabbitmqMsg RabbitMQ消息服务结构体
type sRabbitmqMsg struct{}// New 创建新的RabbitMQ消息服务实例
func New() *sRabbitmqMsg {return &sRabbitmqMsg{}
}// init 初始化函数,在包加载时自动注册RabbitMQ消息服务
func init() {service.RegisterRabbitmqMsg(New())
}// SendMessage 发送普通消息到RabbitMQ
// 参数:
// - ctx: 上下文信息
// - message: 要发送的消息内容
// 返回:
// - error: 发送错误,成功则为nil
func (s *sRabbitmqMsg) SendMessage(ctx context.Context, message string) error {return rabbitmq.PublishMessage(ctx, message)
}// SendDelayMessage 发送延迟消息到RabbitMQ
// 参数:
// - ctx: 上下文信息
// - message: 要发送的消息内容
// - delaySeconds: 延迟时间(秒)
// 返回:
// - error: 发送错误,成功则为nil
func (s *sRabbitmqMsg) SendDelayMessage(ctx context.Context, message string, delaySeconds int) error {return rabbitmq.PublishDelayMessage(ctx, message, delaySeconds)
}// SendBatchMessages 批量发送消息到RabbitMQ
// 参数:
// - ctx: 上下文信息
// - messages: 消息内容数组
// 返回:
// - error: 发送错误,成功则为nil
// 注意:任一消息发送失败都会导致整个批次失败
func (s *sRabbitmqMsg) SendBatchMessages(ctx context.Context, messages []string) error {for _, msg := range messages {if err := rabbitmq.PublishMessage(ctx, msg); err != nil {return err}}return nil
}// GetQueueLength 获取队列当前的消息数量
// 参数:
// - ctx: 上下文信息
// 返回:
// - int: 队列中的消息数量
// - error: 获取错误,成功则为nil
func (s *sRabbitmqMsg) GetQueueLength(ctx context.Context) (int, error) {queue, err := rabbitmq.GetChannel().QueueInspect(g.Cfg().MustGet(ctx, "rabbitmq.queue").String(),)if err != nil {return 0, fmt.Errorf("failed to inspect queue: %v", err)}return queue.Messages, nil
}// PurgeQueue 清空队列中的所有消息
// 参数:
// - ctx: 上下文信息
// 返回:
// - error: 清空错误,成功则为nil
func (s *sRabbitmqMsg) PurgeQueue(ctx context.Context) error {return rabbitmq.PurgeQueue(ctx)
}// handleMessage 处理接收到的单条消息
// 参数:
// - message: 消息内容
// 返回:
// - error: 处理错误,成功则为nil
// 注意:这是内部方法,实现具体的消息处理逻辑
func (s *sRabbitmqMsg) handleMessage(message string) error {// 记录接收到的消息g.Log().Info(context.Background(), "收到消息:", message)// TODO: 在这里添加实际的消息处理逻辑return nil
}// Initialize 初始化消息消费处理
// 参数:
// - ctx: 上下文信息
// 返回:
// - error: 初始化错误,成功则为nil
// 功能:启动消息消费者,并设置消息处理函数
func (s *sRabbitmqMsg) Initialize(ctx context.Context) error {return rabbitmq.ConsumeMessages(ctx, func(msg string) error {return s.handleMessage(msg)})
}
生成service ,不再写上
controller代码
package frontimport ("fmt""gf_new_web/internal/service""time""github.com/gogf/gf/v2/frame/g""github.com/gogf/gf/v2/net/ghttp"
)var (RabbitMsg = cRabbitMsg{}
)type cRabbitMsg struct{}// SendMessage 处理发送普通消息的HTTP请求
// 请求参数:
// - message: 消息内容
// 响应格式:
// 成功:{"code": 0, "msg": "消息发送成功"}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendMessage(r *ghttp.Request) {message := r.Get("message").String()err := service.RabbitmqMsg().SendMessage(r.GetCtx(), message)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": "消息发送成功",})
}// SendDelayMessage 处理发送延迟消息的HTTP请求
// 请求参数:
// - message: 消息内容
// - delay: 延迟时间(秒)
// 响应格式:
// 成功:{"code": 0, "msg": "延迟消息发送成功"}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendDelayMessage(r *ghttp.Request) {message := r.Get("message").String()delaySeconds := r.Get("delay").Int()err := service.RabbitmqMsg().SendDelayMessage(r.GetCtx(), message, delaySeconds)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": "延迟消息发送成功",})
}// SendBatchMessages 处理批量发送消息的HTTP请求
// 请求参数:
// - messages: 消息内容数组
// 响应格式:
// 成功:{"code": 0, "msg": "批量消息发送成功"}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) SendBatchMessages(r *ghttp.Request) {messages := r.Get("messages").Strings()err := service.RabbitmqMsg().SendBatchMessages(r.GetCtx(), messages)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": "批量消息发送成功",})
}// GetQueueLength 处理获取队列长度的HTTP请求
// 响应格式:
// 成功:{"code": 0, "msg": "获取队列长度成功", "data": 队列长度}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) GetQueueLength(r *ghttp.Request) {length, err := service.RabbitmqMsg().GetQueueLength(r.GetCtx())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": "获取队列长度成功","data": length,})
}// PurgeQueue 处理清空队列的HTTP请求
// 响应格式:
// 成功:{"code": 0, "msg": "清空队列成功"}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) PurgeQueue(r *ghttp.Request) {err := service.RabbitmqMsg().PurgeQueue(r.GetCtx())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": "清空队列成功",})
}// ConsumeMessages 处理消费消息的HTTP请求
// 特点:异步处理,非阻塞
// 响应格式:
// 成功:{"code": 0, "msg": "消息消费已开始,请查看服务器日志获取消费详情"}
// 失败:{"code": -1, "msg": "错误信息"}
func (c *cRabbitMsg) ConsumeMessages(r *ghttp.Request) {g.Log().Info(r.GetCtx(), "开始消费消息...")done := make(chan bool)go func() {err := service.RabbitmqMsg().Initialize(r.GetCtx())if err != nil {g.Log().Error(r.GetCtx(), "消费消息出错:", err)r.Response.WriteJson(g.Map{"code": -1,"msg": fmt.Sprintf("消费消息失败: %v", err),})done <- truereturn}}()select {case <-done:returncase <-time.After(5 * time.Second):g.Log().Info(r.GetCtx(), "消息消费进行中...")r.Response.WriteJson(g.Map{"code": 0,"msg": "消息消费已开始,请查看服务器日志获取消费详情",})}
}