package rabbitmq import ( "context" "fmt" "github.com/gogf/gf/v2/frame/g" amqp "github.com/rabbitmq/amqp091-go" ) // QueueConfig 队列配置 type QueueConfig struct { Name string Durable bool // 持久化 AutoDelete bool // 自动删除 Exclusive bool // 排他 Args amqp.Table // 额外参数 } // ExchangeConfig Exchange 配置 type ExchangeConfig struct { Name string Type string // direct/topic/fanout/x-delayed-message Durable bool AutoDelete bool Args amqp.Table } // BindingConfig 绑定配置 type BindingConfig struct { Queue string Exchange string RoutingKey string Args amqp.Table } // DeclareQueue 声明队列 func DeclareQueue(ctx context.Context, cfg *QueueConfig) error { ch, err := GetChannel() if err != nil { return err } _, err = ch.QueueDeclare( cfg.Name, cfg.Durable, cfg.AutoDelete, cfg.Exclusive, false, // no-wait cfg.Args, ) if err != nil { g.Log().Errorf(ctx, "声明队列失败: %s, err=%v", cfg.Name, err) return err } g.Log().Infof(ctx, "队列声明成功: %s", cfg.Name) return nil } // DeclareExchange 声明 Exchange func DeclareExchange(ctx context.Context, cfg *ExchangeConfig) error { ch, err := GetChannel() if err != nil { return err } err = ch.ExchangeDeclare( cfg.Name, cfg.Type, cfg.Durable, cfg.AutoDelete, false, // internal false, // no-wait cfg.Args, ) if err != nil { g.Log().Errorf(ctx, "声明 Exchange 失败: %s, err=%v", cfg.Name, err) return err } g.Log().Infof(ctx, "Exchange 声明成功: %s (type=%s)", cfg.Name, cfg.Type) return nil } // BindQueue 绑定队列到 Exchange func BindQueue(ctx context.Context, cfg *BindingConfig) error { ch, err := GetChannel() if err != nil { return err } err = ch.QueueBind( cfg.Queue, cfg.RoutingKey, cfg.Exchange, false, // no-wait cfg.Args, ) if err != nil { g.Log().Errorf(ctx, "绑定队列失败: queue=%s, exchange=%s, routingKey=%s, err=%v", cfg.Queue, cfg.Exchange, cfg.RoutingKey, err) return err } g.Log().Infof(ctx, "队列绑定成功: queue=%s → exchange=%s (routingKey=%s)", cfg.Queue, cfg.Exchange, cfg.RoutingKey) return nil } // SetupDelayExchange 设置延时 Exchange(需要 rabbitmq_delayed_message_exchange 插件) func SetupDelayExchange(ctx context.Context, exchangeName string) error { return DeclareExchange(ctx, &ExchangeConfig{ Name: exchangeName, Type: "x-delayed-message", Durable: true, Args: amqp.Table{ "x-delayed-type": "direct", }, }) } // SetupDeadLetterQueue 设置死信队列 func SetupDeadLetterQueue(ctx context.Context, queueName, exchangeName string) error { // 1. 声明死信 Exchange err := DeclareExchange(ctx, &ExchangeConfig{ Name: exchangeName, Type: "direct", Durable: true, }) if err != nil { return err } // 2. 声明死信队列 err = DeclareQueue(ctx, &QueueConfig{ Name: queueName, Durable: true, }) if err != nil { return err } // 3. 绑定 return BindQueue(ctx, &BindingConfig{ Queue: queueName, Exchange: exchangeName, RoutingKey: queueName, }) } // SetupQueueWithDLX 创建带死信队列的普通队列 func SetupQueueWithDLX(ctx context.Context, queueName, dlxExchange, dlxRoutingKey string) error { return DeclareQueue(ctx, &QueueConfig{ Name: queueName, Durable: true, Args: amqp.Table{ "x-dead-letter-exchange": dlxExchange, "x-dead-letter-routing-key": dlxRoutingKey, }, }) } // SetupBasicTopology 设置基础拓扑(适用于小红书客服场景) func SetupBasicTopology(ctx context.Context) error { // 1. 声明普通 Exchange err := DeclareExchange(ctx, &ExchangeConfig{ Name: "ragflow_exchange", Type: "direct", Durable: true, }) if err != nil { return err } // 2. 声明延时 Exchange err = SetupDelayExchange(ctx, "delay_exchange") if err != nil { return fmt.Errorf("延时 Exchange 声明失败(可能未安装插件): %v", err) } // 3. 声明死信队列 err = SetupDeadLetterQueue(ctx, "dead_letter_queue", "dlx_exchange") if err != nil { return err } // 4. 声明业务队列 queues := []struct { name string dlx bool // 是否需要死信队列 }{ {"ragflow_request_queue", true}, {"follow_up_queue", true}, {"archive_queue", true}, } for _, q := range queues { if q.dlx { err = SetupQueueWithDLX(ctx, q.name, "dlx_exchange", "dead_letter_queue") } else { err = DeclareQueue(ctx, &QueueConfig{ Name: q.name, Durable: true, }) } if err != nil { return err } } // 5. 绑定队列 bindings := []BindingConfig{ {Queue: "ragflow_request_queue", Exchange: "ragflow_exchange", RoutingKey: "ragflow_request_queue"}, {Queue: "follow_up_queue", Exchange: "delay_exchange", RoutingKey: "follow_up_queue"}, {Queue: "archive_queue", Exchange: "delay_exchange", RoutingKey: "archive_queue"}, } for _, b := range bindings { err = BindQueue(ctx, &b) if err != nil { return err } } g.Log().Info(ctx, "RabbitMQ 拓扑结构设置完成") return nil }