feat(P1): MQ Worker 死信队列 — 重试上限3次后 Ack 移入 DLX
This commit is contained in:
Sisyphus
53
backend/internal/middleware/rabbitmq/dlx.go
Normal file
53
backend/internal/middleware/rabbitmq/dlx.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"log"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
)
|
||||
|
||||
const (
|
||||
DLXExchange = "dlx.events"
|
||||
MaxRetryCount = 3
|
||||
)
|
||||
|
||||
// DeclareDLX 声明死信交换机和对应的死信队列
|
||||
func DeclareDLX(ch *amqp.Channel, queueName string) error {
|
||||
if ch == nil {
|
||||
return nil
|
||||
}
|
||||
if err := ch.ExchangeDeclare(
|
||||
DLXExchange, "topic", true, false, false, false, nil,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
dlxQueue := queueName + ".dlx"
|
||||
_, err := ch.QueueDeclare(
|
||||
dlxQueue, true, false, false, false, nil,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ch.QueueBind(dlxQueue, "#", DLXExchange, false, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("DLX ready: exchange=%s queue=%s", DLXExchange, dlxQueue)
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetRetryCount 从 AMQP x-death header 中提取当前消息已被重试的次数
|
||||
func GetRetryCount(d amqp.Delivery) int {
|
||||
deaths, ok := d.Headers["x-death"].([]interface{})
|
||||
if !ok || len(deaths) == 0 {
|
||||
return 0
|
||||
}
|
||||
death, ok := deaths[0].(amqp.Table)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
count, ok := death["count"].(int64)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
return int(count)
|
||||
}
|
||||
@@ -1,116 +1,123 @@
|
||||
package rabbitmq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"feedsystem_video_go/internal/config"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
)
|
||||
|
||||
type RabbitMQ struct {
|
||||
Conn *amqp.Connection
|
||||
Ch *amqp.Channel
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &RabbitMQ{Conn: conn, Ch: ch}, nil
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) Close() error {
|
||||
if r == nil || r.Ch == nil || r.Conn == nil {
|
||||
return nil
|
||||
}
|
||||
if err := r.Ch.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := r.Conn.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error {
|
||||
if r == nil || r.Ch == nil {
|
||||
return errors.New("rabbitmq is not initialized")
|
||||
}
|
||||
if exchange == "" || queue == "" || bindingKey == "" {
|
||||
return errors.New("exchange/queue/bindingKey is required")
|
||||
}
|
||||
|
||||
if err := r.Ch.ExchangeDeclare(
|
||||
exchange,
|
||||
"topic",
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
nil,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
q, err := r.Ch.QueueDeclare(
|
||||
queue,
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
nil,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return r.Ch.QueueBind(
|
||||
q.Name,
|
||||
bindingKey,
|
||||
exchange,
|
||||
false,
|
||||
nil,
|
||||
)
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error {
|
||||
if r == nil || r.Ch == nil {
|
||||
return errors.New("rabbitmq 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 r.Ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
|
||||
ContentType: "application/json",
|
||||
DeliveryMode: amqp.Persistent,
|
||||
Timestamp: time.Now(),
|
||||
Body: b,
|
||||
})
|
||||
}
|
||||
|
||||
func newEventID(n int) (string, error) {
|
||||
b := make([]byte, n)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(b), nil
|
||||
}
|
||||
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"
|
||||
)
|
||||
|
||||
type RabbitMQ struct {
|
||||
Conn *amqp.Connection
|
||||
Ch *amqp.Channel
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &RabbitMQ{Conn: conn, Ch: ch}, nil
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) Close() error {
|
||||
if r == nil || r.Ch == nil || r.Conn == nil {
|
||||
return nil
|
||||
}
|
||||
if err := r.Ch.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := r.Conn.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error {
|
||||
if r == nil || r.Ch == nil {
|
||||
return errors.New("rabbitmq is not initialized")
|
||||
}
|
||||
if exchange == "" || queue == "" || bindingKey == "" {
|
||||
return errors.New("exchange/queue/bindingKey is required")
|
||||
}
|
||||
|
||||
if err := r.Ch.ExchangeDeclare(
|
||||
exchange,
|
||||
"topic",
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
nil,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
q, err := r.Ch.QueueDeclare(
|
||||
queue,
|
||||
true,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
amqp.Table{"x-dead-letter-exchange": DLXExchange},
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := r.Ch.QueueBind(
|
||||
q.Name,
|
||||
bindingKey,
|
||||
exchange,
|
||||
false,
|
||||
nil,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := DeclareDLX(r.Ch, queue); err != nil {
|
||||
log.Printf("DLX declare failed for %s: %v", queue, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error {
|
||||
if r == nil || r.Ch == nil {
|
||||
return errors.New("rabbitmq 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 r.Ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
|
||||
ContentType: "application/json",
|
||||
DeliveryMode: amqp.Persistent,
|
||||
Timestamp: time.Now(),
|
||||
Body: b,
|
||||
})
|
||||
}
|
||||
|
||||
func newEventID(n int) (string, error) {
|
||||
b := make([]byte, n)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(b), nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user