124 lines
3.0 KiB
Go
124 lines
3.0 KiB
Go
package rabbitmq
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"feedsystem_video_go/internal/config"
|
||
"log"
|
||
"strconv"
|
||
"time"
|
||
|
||
amqp "github.com/rabbitmq/amqp091-go"
|
||
)
|
||
|
||
// RabbitMQ 只管理 Connection，Channel 由各组件按需创建
|
||
type RabbitMQ struct {
|
||
Conn *amqp.Connection
|
||
}
|
||
|
||
func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) {
|
||
if cfg == nil {
|
||
return nil, errors.New("rabbitmq config is nil")
|
||
}
|
||
url := "amqp://" + cfg.Username + ":" + cfg.Password + "@" + cfg.Host + ":" + strconv.Itoa(cfg.Port) + "/"
|
||
conn, err := amqp.Dial(url)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &RabbitMQ{Conn: conn}, nil
|
||
}
|
||
|
||
func (r *RabbitMQ) Close() error {
|
||
if r == nil {
|
||
return nil
|
||
}
|
||
if r.Conn != nil {
|
||
return r.Conn.Close()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *RabbitMQ) NewChannel() (*amqp.Channel, error) {
|
||
if r == nil || r.Conn == nil {
|
||
return nil, errors.New("rabbitmq connection is not initialized")
|
||
}
|
||
return r.Conn.Channel()
|
||
}
|
||
|
||
func DeclareTopic(ch *amqp.Channel, exchange string, queue string, bindingKey string) error {
|
||
if ch == nil {
|
||
return errors.New("channel is not initialized")
|
||
}
|
||
if exchange == "" || queue == "" || bindingKey == "" {
|
||
return errors.New("exchange/queue/bindingKey is required")
|
||
}
|
||
|
||
if err := ch.ExchangeDeclare(
|
||
exchange,
|
||
"topic",
|
||
true, // 持久化参数
|
||
false, // autoDelete 是否在未使用时自动删除
|
||
false, // internal 是否交给其他交换机用而不直接收消息
|
||
false, // noWait 是否不等待 broker 确认
|
||
nil,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
|
||
q, err := ch.QueueDeclare(
|
||
queue, // 队列名称
|
||
true, // 持久化
|
||
false, // autoDelete 是否在未使用时自动删除
|
||
false, // exclusive 是否排他（允许多个消费者共享）
|
||
false, // noWait 是否不等待 broker 确认
|
||
amqp.Table{"x-dead-letter-exchange": DLXExchange}, // 死信交换机
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if err := ch.QueueBind(
|
||
q.Name,
|
||
bindingKey,
|
||
exchange,
|
||
false, // noWait
|
||
nil, // args
|
||
); err != nil {
|
||
return err
|
||
}
|
||
if err := DeclareDLX(ch, queue); err != nil {
|
||
log.Printf("DLX declare failed for %s: %v", queue, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func PublishJSON(ctx context.Context, ch *amqp.Channel, exchange string, routingKey string, payload any) error {
|
||
if ch == nil {
|
||
return errors.New("channel is not initialized")
|
||
}
|
||
if exchange == "" || routingKey == "" {
|
||
return errors.New("exchange and routingKey are required")
|
||
}
|
||
b, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
|
||
ContentType: "application/json", // 消息格式
|
||
DeliveryMode: amqp.Persistent, // 消息持久化到磁盘, 值为 1 则不持久化, 值为 2 则持久化, amqp.Presistent 为 2
|
||
Timestamp: time.Now(), // 消息产生的时间戳
|
||
Body: b, // 实际消息内容(JSON字节)
|
||
})
|
||
}
|
||
|
||
func newEventID(n int) (string, error) {
|
||
b := make([]byte, n)
|
||
if _, err := rand.Read(b); err != nil {
|
||
return "", err
|
||
}
|
||
return hex.EncodeToString(b), nil
|
||
}
|