goframe开发一个企业网站 rabbitmq队例15

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":  "消息消费已开始,请查看服务器日志获取消费详情",})}
}