package rabbitmq import ( "context" "encoding/json" "sync" "time" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" amqp "github.com/rabbitmq/amqp091-go" ) type Config struct { Enabled bool Dsn string } var ( conn *amqp.Connection connMu sync.Mutex ) func GetConfig(ctx context.Context) Config { return Config{ Enabled: g.Cfg().MustGet(ctx, "rabbitmq.enabled", false).Bool(), Dsn: g.Cfg().MustGet(ctx, "rabbitmq.dsn", "").String(), } } func Enabled(ctx context.Context) bool { cfg := GetConfig(ctx) return cfg.Enabled && cfg.Dsn != "" } func Publish(ctx context.Context, queue string, data any) error { body, err := json.Marshal(data) if err != nil { return gerror.Wrap(err, "序列化 RabbitMQ 消息失败") } return PublishBytes(ctx, queue, body) } func PublishBytes(ctx context.Context, queue string, body []byte) error { if queue == "" { return gerror.New("RabbitMQ 队列名不能为空") } ch, err := channel(ctx) if err != nil { return err } defer ch.Close() if _, err = ch.QueueDeclare(queue, true, false, false, false, nil); err != nil { return gerror.Wrapf(err, "声明 RabbitMQ 队列失败: %s", queue) } return ch.PublishWithContext(ctx, "", queue, false, false, amqp.Publishing{ ContentType: "application/json", DeliveryMode: amqp.Persistent, Timestamp: time.Now(), Body: body, }) } func channel(ctx context.Context) (*amqp.Channel, error) { c, err := connection(ctx) if err != nil { return nil, err } ch, err := c.Channel() if err != nil { return nil, gerror.Wrap(err, "打开 RabbitMQ Channel 失败") } return ch, nil } func connection(ctx context.Context) (*amqp.Connection, error) { cfg := GetConfig(ctx) if !cfg.Enabled { return nil, gerror.New("RabbitMQ 未启用") } if cfg.Dsn == "" { return nil, gerror.New("RabbitMQ DSN 未配置") } connMu.Lock() defer connMu.Unlock() if conn != nil && !conn.IsClosed() { return conn, nil } c, err := amqp.Dial(cfg.Dsn) if err != nil { return nil, gerror.Wrap(err, "连接 RabbitMQ 失败") } conn = c return conn, nil }