From d8bca1636271545c194a50be49a3297635985c23 Mon Sep 17 00:00:00 2001 From: yiyiis Date: Fri, 22 May 2026 23:57:06 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8B=86=E5=88=86=20RabbitMQ=20Channel?= =?UTF-8?q?=EF=BC=8C=E4=BF=AE=E5=A4=8D=E6=96=B0=E8=A7=86=E9=A2=91=E4=B8=8D?= =?UTF-8?q?=E5=87=BA=E7=8E=B0=E5=9C=A8=E6=8E=A8=E8=8D=90=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RabbitMQ 结构体移除共享 Ch 字段，仅管理 Connection - 每个MQ组件(Like/Comment/Social/Popularity/Timeline)持有独立 Channel - Timeline Consumer 使用独立 Channel 并加入自动重连机制 - Redis 操作超时从 50ms 调整为 500ms，避免 NACK 循环 - router.go 中 notification 相关逻辑适配新 API --- backend/internal/http/router.go | 31 +++--- .../internal/middleware/rabbitmq/commentMQ.go | 17 ++- .../internal/middleware/rabbitmq/likeMQ.go | 17 ++- .../middleware/rabbitmq/popularityMQ.go | 17 ++- .../internal/middleware/rabbitmq/rabbitMQ.go | 50 ++++----- .../internal/middleware/rabbitmq/socialMQ.go | 17 ++- .../middleware/rabbitmq/timelineMQ.go | 17 ++- backend/internal/worker/outboxworker.go | 104 ++++++++++-------- 8 files changed, 156 insertions(+), 114 deletions(-) diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index df6f420..c619935 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -201,18 +201,21 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g timelineMQ = nil } worker.StartOutboxPoller(db, timelineMQ) - worker.StartConsumer(timelineMQ, "video.timeline.update.queue", cache) + worker.StartConsumer(timelineMQ, "video.timeline.update.queue", cache, rmq) // SSE notification - if rmq != nil && rmq.Ch != nil { - if err := rmq.DeclareTopic("like.events", "notification.like", "like.like"); err != nil { - log.Printf("notification like topic init failed: %v", err) - } - if err := rmq.DeclareTopic("comment.events", "notification.comment", "comment.publish"); err != nil { - log.Printf("notification comment topic init failed: %v", err) - } - if err := rmq.DeclareTopic("social.events", "notification.social", "social.follow"); err != nil { - log.Printf("notification social topic init failed: %v", err) + if rmq != nil { + if notifCh, err := rmq.NewChannel(); err == nil { + if err := rabbitmq.DeclareTopic(notifCh, "like.events", "notification.like", "like.like"); err != nil { + log.Printf("notification like topic init failed: %v", err) + } + if err := rabbitmq.DeclareTopic(notifCh, "comment.events", "notification.comment", "comment.publish"); err != nil { + log.Printf("notification comment topic init failed: %v", err) + } + if err := rabbitmq.DeclareTopic(notifCh, "social.events", "notification.social", "social.follow"); err != nil { + log.Printf("notification social topic init failed: %v", err) + } + notifCh.Close() } } sseHub := worker.NewSSEHub(db) @@ -221,12 +224,12 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g sseHub.RegisterRoutes(r, notifGroup) go func() { - if rmq != nil && rmq.Ch != nil { + if rmq != nil { hub := sseHub ctx := context.Background() // consume from like queue go func() { - ch, err := rmq.Conn.Channel() + ch, err := rmq.NewChannel() if err != nil { log.Printf("notification-like channel: %v", err) return @@ -238,7 +241,7 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g } }() go func() { - ch, err := rmq.Conn.Channel() + ch, err := rmq.NewChannel() if err != nil { log.Printf("notification-comment channel: %v", err) return @@ -250,7 +253,7 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g } }() go func() { - ch, err := rmq.Conn.Channel() + ch, err := rmq.NewChannel() if err != nil { log.Printf("notification-social channel: %v", err) return diff --git a/backend/internal/middleware/rabbitmq/commentMQ.go b/backend/internal/middleware/rabbitmq/commentMQ.go index 33ca14c..d8d82ff 100644 --- a/backend/internal/middleware/rabbitmq/commentMQ.go +++ b/backend/internal/middleware/rabbitmq/commentMQ.go @@ -4,10 +4,12 @@ import ( "context" "errors" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type CommentMQ struct { - *RabbitMQ + ch *amqp.Channel } const ( @@ -34,10 +36,15 @@ func NewCommentMQ(base *RabbitMQ) (*CommentMQ, error) { if base == nil { return nil, errors.New("rabbitmq base is nil") } - if err := base.DeclareTopic(commentExchange, commentQueue, commentBindingKey); err != nil { + ch, err := base.NewChannel() + if err != nil { return nil, err } - return &CommentMQ{RabbitMQ: base}, nil + if err := DeclareTopic(ch, commentExchange, commentQueue, commentBindingKey); err != nil { + ch.Close() + return nil, err + } + return &CommentMQ{ch: ch}, nil } func (c *CommentMQ) Publish(ctx context.Context, username string, videoID, authorID uint, content string) error { @@ -56,7 +63,7 @@ func (c *CommentMQ) Delete(ctx context.Context, commentID uint) error { } func (c *CommentMQ) publish(ctx context.Context, action, routingKey string, evt CommentEvent) error { - if c == nil || c.RabbitMQ == nil { + if c == nil || c.ch == nil { return errors.New("comment mq is not initialized") } id, err := newEventID(16) @@ -66,5 +73,5 @@ func (c *CommentMQ) publish(ctx context.Context, action, routingKey string, evt evt.EventID = id evt.Action = action evt.OccurredAt = time.Now().UTC() - return c.PublishJSON(ctx, commentExchange, routingKey, evt) + return PublishJSON(ctx, c.ch, commentExchange, routingKey, evt) } diff --git a/backend/internal/middleware/rabbitmq/likeMQ.go b/backend/internal/middleware/rabbitmq/likeMQ.go index f9b1d30..6bfc65d 100644 --- a/backend/internal/middleware/rabbitmq/likeMQ.go +++ b/backend/internal/middleware/rabbitmq/likeMQ.go @@ -4,10 +4,12 @@ import ( "context" "errors" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type LikeMQ struct { - *RabbitMQ + ch *amqp.Channel } const ( @@ -31,10 +33,15 @@ func NewLikeMQ(base *RabbitMQ) (*LikeMQ, error) { if base == nil { return nil, errors.New("rabbitmq base is nil") } - if err := base.DeclareTopic(likeExchange, likeQueue, likeBindingKey); err != nil { + ch, err := base.NewChannel() + if err != nil { return nil, err } - return &LikeMQ{RabbitMQ: base}, nil + if err := DeclareTopic(ch, likeExchange, likeQueue, likeBindingKey); err != nil { + ch.Close() + return nil, err + } + return &LikeMQ{ch: ch}, nil } func (l *LikeMQ) Like(ctx context.Context, userID, videoID uint) error { @@ -46,7 +53,7 @@ func (l *LikeMQ) Unlike(ctx context.Context, userID, videoID uint) error { } func (l *LikeMQ) publish(ctx context.Context, action, routingKey string, userID, videoID uint) error { - if l == nil || l.RabbitMQ == nil { + if l == nil || l.ch == nil { return errors.New("like mq is not initialized") } if userID == 0 || videoID == 0 { @@ -63,5 +70,5 @@ func (l *LikeMQ) publish(ctx context.Context, action, routingKey string, userID, VideoID: videoID, OccurredAt: time.Now(), } - return l.PublishJSON(ctx, likeExchange, routingKey, event) + return PublishJSON(ctx, l.ch, likeExchange, routingKey, event) } diff --git a/backend/internal/middleware/rabbitmq/popularityMQ.go b/backend/internal/middleware/rabbitmq/popularityMQ.go index 572d2f1..e829c78 100644 --- a/backend/internal/middleware/rabbitmq/popularityMQ.go +++ b/backend/internal/middleware/rabbitmq/popularityMQ.go @@ -4,10 +4,12 @@ import ( "context" "errors" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type PopularityMQ struct { - *RabbitMQ + ch *amqp.Channel } const ( @@ -29,14 +31,19 @@ func NewPopularityMQ(base *RabbitMQ) (*PopularityMQ, error) { if base == nil { return nil, errors.New("rabbitmq base is nil") } - if err := base.DeclareTopic(popularityExchange, popularityQueue, popularityBindingKey); err != nil { + ch, err := base.NewChannel() + if err != nil { return nil, err } - return &PopularityMQ{RabbitMQ: base}, nil + if err := DeclareTopic(ch, popularityExchange, popularityQueue, popularityBindingKey); err != nil { + ch.Close() + return nil, err + } + return &PopularityMQ{ch: ch}, nil } func (p *PopularityMQ) Update(ctx context.Context, videoID uint, change int64) error { - if p == nil || p.RabbitMQ == nil { + if p == nil || p.ch == nil { return errors.New("popularity mq is not initialized") } if videoID == 0 || change == 0 { @@ -52,5 +59,5 @@ func (p *PopularityMQ) Update(ctx context.Context, videoID uint, change int64) e Change: change, OccurredAt: time.Now().UTC(), } - return p.PublishJSON(ctx, popularityExchange, popularityUpdateRK, event) + return PublishJSON(ctx, p.ch, popularityExchange, popularityUpdateRK, event) } diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go index 1b8eeda..63b3616 100644 --- a/backend/internal/middleware/rabbitmq/rabbitMQ.go +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -14,9 +14,9 @@ import ( amqp "github.com/rabbitmq/amqp091-go" ) +// RabbitMQ 只管理 Connection，Channel 由各组件按需创建 type RabbitMQ struct { Conn *amqp.Connection - Ch *amqp.Channel } func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { @@ -28,41 +28,35 @@ func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { if err != nil { return nil, err } - ch, err := conn.Channel() - if err != nil { - _ = conn.Close() - return nil, err - } - return &RabbitMQ{Conn: conn, Ch: ch}, nil + return &RabbitMQ{Conn: conn}, nil } func (r *RabbitMQ) Close() error { if r == nil { return nil } - var closeErr error - if r.Ch != nil { - if err := r.Ch.Close(); err != nil { - closeErr = err - } - } if r.Conn != nil { - if err := r.Conn.Close(); closeErr == nil && err != nil { - closeErr = err - } + return r.Conn.Close() } - return closeErr + 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") +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 := r.Ch.ExchangeDeclare( + if err := ch.ExchangeDeclare( exchange, "topic", true, @@ -74,7 +68,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string return err } - q, err := r.Ch.QueueDeclare( + q, err := ch.QueueDeclare( queue, true, false, @@ -86,7 +80,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string return err } - if err := r.Ch.QueueBind( + if err := ch.QueueBind( q.Name, bindingKey, exchange, @@ -95,15 +89,15 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string ); err != nil { return err } - if err := DeclareDLX(r.Ch, queue); err != nil { + if err := DeclareDLX(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") +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") @@ -112,7 +106,7 @@ func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey if err != nil { return err } - return r.Ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ + return ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ ContentType: "application/json", DeliveryMode: amqp.Persistent, Timestamp: time.Now(), diff --git a/backend/internal/middleware/rabbitmq/socialMQ.go b/backend/internal/middleware/rabbitmq/socialMQ.go index 40b1ad2..b8df788 100644 --- a/backend/internal/middleware/rabbitmq/socialMQ.go +++ b/backend/internal/middleware/rabbitmq/socialMQ.go @@ -4,10 +4,12 @@ import ( "context" "errors" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type SocialMQ struct { - *RabbitMQ + ch *amqp.Channel } const ( @@ -31,10 +33,15 @@ func NewSocialMQ(base *RabbitMQ) (*SocialMQ, error) { if base == nil { return nil, errors.New("rabbitmq base is nil") } - if err := base.DeclareTopic(socialExchange, socialQueue, socialBindingKey); err != nil { + ch, err := base.NewChannel() + if err != nil { return nil, err } - return &SocialMQ{RabbitMQ: base}, nil + if err := DeclareTopic(ch, socialExchange, socialQueue, socialBindingKey); err != nil { + ch.Close() + return nil, err + } + return &SocialMQ{ch: ch}, nil } func (s *SocialMQ) Follow(ctx context.Context, followerID, vloggerID uint) error { @@ -46,7 +53,7 @@ func (s *SocialMQ) UnFollow(ctx context.Context, followerID, vloggerID uint) err } func (s *SocialMQ) publish(ctx context.Context, action, routingKey string, followerID, vloggerID uint) error { - if s == nil || s.RabbitMQ == nil { + if s == nil || s.ch == nil { return errors.New("social mq is not initialized") } if followerID == 0 || vloggerID == 0 { @@ -63,5 +70,5 @@ func (s *SocialMQ) publish(ctx context.Context, action, routingKey string, follo VloggerID: vloggerID, OccurredAt: time.Now().UTC(), } - return s.PublishJSON(ctx, socialExchange, routingKey, evt) + return PublishJSON(ctx, s.ch, socialExchange, routingKey, evt) } diff --git a/backend/internal/middleware/rabbitmq/timelineMQ.go b/backend/internal/middleware/rabbitmq/timelineMQ.go index 898ed0a..f25320a 100644 --- a/backend/internal/middleware/rabbitmq/timelineMQ.go +++ b/backend/internal/middleware/rabbitmq/timelineMQ.go @@ -4,10 +4,12 @@ import ( "context" "errors" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type TimelineMQ struct { - *RabbitMQ + ch *amqp.Channel } const ( @@ -28,14 +30,19 @@ func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) { if base == nil { return nil, errors.New("rabbitmq base is nil") } - if err := base.DeclareTopic(timelineExchange, timelineQueue, timelineBindingKey); err != nil { + ch, err := base.NewChannel() + if err != nil { return nil, err } - return &TimelineMQ{RabbitMQ: base}, nil + if err := DeclareTopic(ch, timelineExchange, timelineQueue, timelineBindingKey); err != nil { + ch.Close() + return nil, err + } + return &TimelineMQ{ch: ch}, nil } func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error { - if t == nil || t.RabbitMQ == nil { + if t == nil || t.ch == nil { return errors.New("timeline mq is not initialized") } if videoID == 0 { @@ -51,5 +58,5 @@ func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime CreateTime: createTime.UnixMilli(), OccurredAt: time.Now(), } - return t.PublishJSON(ctx, timelineExchange, timelinePublishRK, timeline) + return PublishJSON(ctx, t.ch, timelineExchange, timelinePublishRK, timeline) } diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index 1e8ced0..aa87472 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -4,7 +4,7 @@ import ( "context" "encoding/json" "feedsystem_video_go/internal/middleware/rabbitmq" - "feedsystem_video_go/internal/middleware/redis" + rediscache "feedsystem_video_go/internal/middleware/redis" "feedsystem_video_go/internal/video" "fmt" "log" @@ -15,7 +15,7 @@ import ( ) func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { - if db == nil || tmq == nil || tmq.RabbitMQ == nil || tmq.Ch == nil { + if db == nil || tmq == nil { log.Printf("Outbox poller disabled: timeline mq is not initialized") return } @@ -46,9 +46,9 @@ func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { }() } -func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) { - if tmq == nil || tmq.RabbitMQ == nil || tmq.Ch == nil { - log.Printf("Timeline consumer disabled: timeline mq is not initialized") +func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *rediscache.Client, rmq *rabbitmq.RabbitMQ) { + if tmq == nil || rmq == nil || rmq.Conn == nil { + log.Printf("Timeline consumer disabled: rabbitmq is not initialized") return } if redisClient == nil { @@ -56,54 +56,64 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi return } - msgs, err := tmq.Ch.Consume( - queueName, - "", - false, - false, - false, - false, - nil, - ) - - if err != nil { - log.Printf("注册消费失败") - return - } - go func() { - for msg := range msgs { - var event rabbitmq.TimelineEvent - err := json.Unmarshal(msg.Body, &event) - + for { + // 每次重连创建独立的 Channel，不与发布者共用 + ch, err := rmq.NewChannel() if err != nil { - log.Printf("反序列化失败") + log.Printf("Timeline consumer: 创建 Channel 失败: %v, 5秒后重试", err) + time.Sleep(5 * time.Second) + continue + } + + if err := ch.Qos(10, 0, false); err != nil { + log.Printf("Timeline consumer: QoS 设置失败: %v", err) + } + + msgs, err := ch.Consume(queueName, "", false, false, false, false, nil) + if err != nil { + log.Printf("Timeline consumer: 注册消费失败: %v, 5秒后重试", err) + ch.Close() + time.Sleep(5 * time.Second) + continue + } + + log.Printf("Timeline consumer 已启动, queue=%s", queueName) + + for msg := range msgs { + var event rabbitmq.TimelineEvent + if err := json.Unmarshal(msg.Body, &event); err != nil { + log.Printf("Timeline consumer: 反序列化失败: %v", err) + msg.Ack(false) + continue + } + + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + timelineKey := redisClient.Key("feed:global_timeline") + err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ + Score: float64(event.CreateTime), + Member: fmt.Sprintf("%d", event.VideoID), + }) + + if err != nil { + log.Printf("Timeline consumer: 写入Zset失败: %v", err) + msg.Nack(false, true) + cancel() + continue + } + + if err := redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001); err != nil { + log.Printf("Timeline consumer: ZRem失败: %v", err) + } + msg.Ack(false) - continue - } - - ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) - timelineKey := redisClient.Key("feed:global_timeline") - err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ - Score: float64(event.CreateTime), - Member: fmt.Sprintf("%d", event.VideoID), - }) - - if err != nil { - log.Printf("写入Zset失败") - msg.Nack(false, true) cancel() - continue } - err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001) - - if err != nil { - log.Printf("ZRem失败") - } - - msg.Ack(false) - cancel() + // msgs channel 关闭说明 AMQP Channel 断开，关闭并重连 + ch.Close() + log.Printf("Timeline consumer: Channel 断开, 5秒后重连...") + time.Sleep(5 * time.Second) } }() }