huanghaosheng/VLoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 拆分 RabbitMQ Channel，修复新视频不出现在推荐列表的问题

Browse Source 
- RabbitMQ 结构体移除共享 Ch 字段，仅管理 Connection
- 每个MQ组件(Like/Comment/Social/Popularity/Timeline)持有独立 Channel
- Timeline Consumer 使用独立 Channel 并加入自动重连机制
- Redis 操作超时从 50ms 调整为 500ms，避免 NACK 循环
- router.go 中 notification 相关逻辑适配新 API
This commit is contained in:
yiyiis
2026-05-22 23:57:06 +08:00
parent 229446e93b
commit d8bca16362
8 changed files with 156 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
backend/internal/http/router.go
View File

@@ -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