From cbfdb73fcbd49ac07aa5275128a8d72b6b3339a1 Mon Sep 17 00:00:00 2001 From: yiyiis Date: Sat, 23 May 2026 09:23:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20Notification=20=E9=87=8D=E8=BF=9E=20+=20?= =?UTF-8?q?Social=20=E4=BF=AE=E5=A4=8D=20+=20Following=20=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E5=A4=B1=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Notification Worker 加重连循环，Channel 断开不再静默死亡 - Social: 先写 DB 再发 MQ，修复 DB 失败产生幽灵通知的问题 - Social: MQ 发布失败记日志，不再静默忽略 - 关注/取关后失效该用户的 Following Feed 缓存（24h TTL） - Redis Client 新增 DelByPattern 按模式批量删除缓存 --- backend/internal/http/router.go | 58 ++++++++-------------- backend/internal/middleware/redis/cache.go | 11 ++++ backend/internal/social/service.go | 53 +++++++++++++++++--- 3 files changed, 76 insertions(+), 46 deletions(-) diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index c619935..5eeaf21 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -127,7 +127,7 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g socialMQ = nil } socialRepository := social.NewSocialRepository(db) - socialService := social.NewSocialService(socialRepository, accountRepository, socialMQ) + socialService := social.NewSocialService(socialRepository, accountRepository, socialMQ, cache) socialHandler := social.NewSocialHandler(socialService) socialGroup := r.Group("/social") protectedSocialGroup := socialGroup.Group("") @@ -227,43 +227,25 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g if rmq != nil { hub := sseHub ctx := context.Background() - // consume from like queue - go func() { - ch, err := rmq.NewChannel() - if err != nil { - log.Printf("notification-like channel: %v", err) - return - } - defer ch.Close() - w := worker.NewNotificationWorker(ch, db, "notification.like", hub) - if err := w.Run(ctx); err != nil { - log.Printf("notification-like worker: %v", err) - } - }() - go func() { - ch, err := rmq.NewChannel() - if err != nil { - log.Printf("notification-comment channel: %v", err) - return - } - defer ch.Close() - w := worker.NewNotificationWorker(ch, db, "notification.comment", hub) - if err := w.Run(ctx); err != nil { - log.Printf("notification-comment worker: %v", err) - } - }() - go func() { - ch, err := rmq.NewChannel() - if err != nil { - log.Printf("notification-social channel: %v", err) - return - } - defer ch.Close() - w := worker.NewNotificationWorker(ch, db, "notification.social", hub) - if err := w.Run(ctx); err != nil { - log.Printf("notification-social worker: %v", err) - } - }() + // 每个 notification worker 独立 Channel + 自动重连 + for _, q := range []string{"notification.like", "notification.comment", "notification.social"} { + go func(queue string) { + for { + ch, err := rmq.NewChannel() + if err != nil { + log.Printf("notification-%s: 创建 Channel 失败: %v, 5秒后重试", queue, err) + time.Sleep(5 * time.Second) + continue + } + w := worker.NewNotificationWorker(ch, db, queue, hub) + if err := w.Run(ctx); err != nil { + log.Printf("notification-%s: %v, 5秒后重连...", queue, err) + } + ch.Close() + time.Sleep(5 * time.Second) + } + }(q) + } } else { log.Printf("Notification SSE disabled (MQ not available)") } diff --git a/backend/internal/middleware/redis/cache.go b/backend/internal/middleware/redis/cache.go index b3e328c..e0369e3 100644 --- a/backend/internal/middleware/redis/cache.go +++ b/backend/internal/middleware/redis/cache.go @@ -27,6 +27,17 @@ func (c *Client) Del(ctx context.Context, key string) error { return c.rdb.Del(ctx, key).Err() } +func (c *Client) DelByPattern(ctx context.Context, pattern string) error { + if c == nil || c.rdb == nil { + return nil + } + iter := c.rdb.Scan(ctx, 0, pattern, 0).Iterator() + for iter.Next(ctx) { + _ = c.rdb.Del(ctx, iter.Val()) + } + return iter.Err() +} + func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) ([]interface{}, error) { if c == nil || c.rdb == nil { return nil, errors.New("redis client not initialized") diff --git a/backend/internal/social/service.go b/backend/internal/social/service.go index dfb6147..e7a9b5c 100644 --- a/backend/internal/social/service.go +++ b/backend/internal/social/service.go @@ -5,16 +5,19 @@ import ( "errors" "feedsystem_video_go/internal/account" "feedsystem_video_go/internal/middleware/rabbitmq" + rediscache "feedsystem_video_go/internal/middleware/redis" + "log" ) type SocialService struct { repo *SocialRepository accountrepo *account.AccountRepository socialMQ *rabbitmq.SocialMQ + cache *rediscache.Client } -func NewSocialService(repo *SocialRepository, accountrepo *account.AccountRepository, socialMQ *rabbitmq.SocialMQ) *SocialService { - return &SocialService{repo: repo, accountrepo: accountrepo, socialMQ: socialMQ} +func NewSocialService(repo *SocialRepository, accountrepo *account.AccountRepository, socialMQ *rabbitmq.SocialMQ, cache *rediscache.Client) *SocialService { + return &SocialService{repo: repo, accountrepo: accountrepo, socialMQ: socialMQ, cache: cache} } func (s *SocialService) Follow(ctx context.Context, social *Social) error { @@ -36,10 +39,22 @@ func (s *SocialService) Follow(ctx context.Context, social *Social) error { if isFollowed { return errors.New("already followed") } - if s.socialMQ != nil { - s.socialMQ.Follow(ctx, social.FollowerID, social.VloggerID) + + // 先写 DB，确保数据持久化 + if err := s.repo.Follow(ctx, social); err != nil { + return err } - return s.repo.Follow(ctx, social) + + // DB 成功后，失效该用户的关注列表缓存 + s.invalidateFollowingFeedCache(context.Background(), social.FollowerID) + + // 最后发 MQ（用于通知），失败只记日志不影响业务 + if s.socialMQ != nil { + if err := s.socialMQ.Follow(ctx, social.FollowerID, social.VloggerID); err != nil { + log.Printf("social MQ Follow 发布失败: %v", err) + } + } + return nil } func (s *SocialService) Unfollow(ctx context.Context, social *Social) error { @@ -58,10 +73,32 @@ func (s *SocialService) Unfollow(ctx context.Context, social *Social) error { if !isFollowed { return errors.New("not followed") } - if s.socialMQ != nil { - s.socialMQ.UnFollow(ctx, social.FollowerID, social.VloggerID) + + // 先写 DB + if err := s.repo.Unfollow(ctx, social); err != nil { + return err + } + + // 失效缓存 + s.invalidateFollowingFeedCache(context.Background(), social.FollowerID) + + // 最后发 MQ + if s.socialMQ != nil { + if err := s.socialMQ.UnFollow(ctx, social.FollowerID, social.VloggerID); err != nil { + log.Printf("social MQ UnFollow 发布失败: %v", err) + } + } + return nil +} + +func (s *SocialService) invalidateFollowingFeedCache(ctx context.Context, accountID uint) { + if s.cache == nil { + return + } + pattern := s.cache.Key("feed:listByFollowing:*:accountID=%d:*", accountID) + if err := s.cache.DelByPattern(ctx, pattern); err != nil { + log.Printf("失效 Following 缓存失败: accountID=%d, err=%v", accountID, err) } - return s.repo.Unfollow(ctx, social) } func (s *SocialService) GetAllFollowers(ctx context.Context, VloggerID uint) ([]*account.Account, error) {