Merge pull request #14 from yiyiis/fix/system-reliability

Fix/system reliability

backend/cmd/worker/main.go

@@ -55,6 +55,38 @@ func connectWithRetry(name string, maxRetries int, fn func() error) {
 	log.Fatalf("%s: 超过最大重试次数", name)
 }
 
+// runWorkerWithRetry 为每个 Worker 创建独立 Channel，断开后自动重连
+func runWorkerWithRetry(ctx context.Context, name string, conn *amqp.Connection, fn func(*amqp.Channel) error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		ch, err := conn.Channel()
+		if err != nil {
+			log.Printf("%s: 创建 Channel 失败: %v, 5秒后重试", name, err)
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		if err := ch.Qos(50, 0, false); err != nil {
+			log.Printf("%s: QoS 设置失败: %v", name, err)
+		}
+
+		log.Printf("%s started, consuming", name)
+		if err := fn(ch); err != nil {
+			if ctx.Err() != nil {
+				ch.Close()
+				return
+			}
+			log.Printf("%s: %v, 5秒后重连...", name, err)
+		}
+		ch.Close()
+		time.Sleep(5 * time.Second)
+	}
+}
+
 func main() {
 	// 加载 .env（本地开发）
 	if err := godotenv.Load(); err != nil {
@@ -110,42 +142,33 @@ func main() {
 		return err
 	})
 	defer conn.Close()
-	// 创建 RabbitMQ 通道
-	ch, err := conn.Channel()
+
+	// 用临时 Channel 声明拓扑（持久化队列，声明一次即可）
+	topoCh, err := conn.Channel()
 	if err != nil {
-		log.Fatalf("Failed to open rabbitmq channel: %v", err)
+		log.Fatalf("Failed to open topology channel: %v", err)
 	}
-	defer ch.Close()
-	// 声明 Social 交换机和队列
-	if err := declareSocialTopology(ch); err != nil {
+	if err := declareSocialTopology(topoCh); err != nil {
 		log.Fatalf("Failed to declare social topology: %v", err)
 	}
-	if err := declareLikeTopology(ch); err != nil {
+	if err := declareLikeTopology(topoCh); err != nil {
 		log.Fatalf("Failed to declare like topology: %v", err)
 	}
-	if err := declareCommentTopology(ch); err != nil {
+	if err := declareCommentTopology(topoCh); err != nil {
 		log.Fatalf("Failed to declare comment topology: %v", err)
 	}
 	if cache != nil {
-		if err := declarePopularityTopology(ch); err != nil {
+		if err := declarePopularityTopology(topoCh); err != nil {
 			log.Fatalf("Failed to declare popularity topology: %v", err)
 		}
 	}
-	if err := ch.Qos(50, 0, false); err != nil {
-		log.Fatalf("Failed to set qos: %v", err)
-	}
+	topoCh.Close()
 
-	repo := social.NewSocialRepository(sqlDB)
-	socialWorker := worker.NewSocialWorker(ch, repo, socialQueue)
+	// 准备 repo
+	socialRepo := social.NewSocialRepository(sqlDB)
 	videoRepo := video.NewVideoRepository(sqlDB)
 	likeRepo := video.NewLikeRepository(sqlDB)
 	commentRepo := video.NewCommentRepository(sqlDB)
-	likeWorker := worker.NewLikeWorker(ch, likeRepo, videoRepo, likeQueue)
-	commentWorker := worker.NewCommentWorker(ch, commentRepo, videoRepo, commentQueue)
-	var popularityWorker *worker.PopularityWorker
-	if cache != nil {
-		popularityWorker = worker.NewPopularityWorker(ch, cache, popularityQueue)
-	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
@@ -162,22 +185,26 @@ func main() {
 		defer pprofServer.Close()
 	}
 
-	errCh := make(chan error, 4)
-	log.Printf("Worker started, consuming queue=%s", socialQueue)
-	go func() { errCh <- socialWorker.Run(ctx) }()
-	log.Printf("Worker started, consuming queue=%s", likeQueue)
-	go func() { errCh <- likeWorker.Run(ctx) }()
-	log.Printf("Worker started, consuming queue=%s", commentQueue)
-	go func() { errCh <- commentWorker.Run(ctx) }()
-	if popularityWorker != nil {
-		log.Printf("Worker started, consuming queue=%s", popularityQueue)
-		go func() { errCh <- popularityWorker.Run(ctx) }()
+	// 每个 Worker 独立 Channel + 自动重连
+	go runWorkerWithRetry(ctx, "SocialWorker", conn, func(ch *amqp.Channel) error {
+		return worker.NewSocialWorker(ch, socialRepo, socialQueue).Run(ctx)
+	})
+	go runWorkerWithRetry(ctx, "LikeWorker", conn, func(ch *amqp.Channel) error {
+		return worker.NewLikeWorker(ch, likeRepo, videoRepo, likeQueue).Run(ctx)
+	})
+	go runWorkerWithRetry(ctx, "CommentWorker", conn, func(ch *amqp.Channel) error {
+		return worker.NewCommentWorker(ch, commentRepo, videoRepo, commentQueue).Run(ctx)
+	})
+	if cache != nil {
+		go runWorkerWithRetry(ctx, "PopularityWorker", conn, func(ch *amqp.Channel) error {
+			return worker.NewPopularityWorker(ch, cache, popularityQueue).Run(ctx)
+		})
 	}
 
-	err = <-errCh
-	if err != nil && err != context.Canceled {
-		log.Fatalf("Worker stopped: %v", err)
-	}
+	// 等待退出信号
+	<-ctx.Done()
+	log.Printf("Worker shutting down...")
+	time.Sleep(2 * time.Second) // 等待正在处理的消息完成
 	log.Printf("Worker stopped")
 }
 

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() {
+			// 每个 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-like channel: %v", err)
-					return
+							log.Printf("notification-%s: 创建 Channel 失败: %v, 5秒后重试", queue, err)
+							time.Sleep(5 * time.Second)
+							continue
 						}
-				defer ch.Close()
-				w := worker.NewNotificationWorker(ch, db, "notification.like", hub)
+						w := worker.NewNotificationWorker(ch, db, queue, hub)
 						if err := w.Run(ctx); err != nil {
-					log.Printf("notification-like worker: %v", err)
+							log.Printf("notification-%s: %v, 5秒后重连...", queue, err)
 						}
-			}()
-			go func() {
-				ch, err := rmq.NewChannel()
-				if err != nil {
-					log.Printf("notification-comment channel: %v", err)
-					return
+						ch.Close()
+						time.Sleep(5 * time.Second)
 					}
-				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)
+				}(q)
 			}
-			}()
-			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)
-				}
-			}()
 		} else {
 			log.Printf("Notification SSE disabled (MQ not available)")
 		}

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")

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) {

backend/internal/worker/commentworker.go

@@ -8,6 +8,7 @@ import (
 	"feedsystem_video_go/internal/video"
 	"log"
 	"strings"
+	"time"
 
 	amqp "github.com/rabbitmq/amqp091-go"
 )
@@ -58,18 +59,28 @@ func (w *CommentWorker) Run(ctx context.Context) error {
 }
 
 func (w *CommentWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
-	if err := w.process(ctx, d.Body); err != nil {
-		retryCount := rabbitmq.GetRetryCount(d)
-		if retryCount >= rabbitmq.MaxRetryCount {
-			log.Printf("comment worker: max retries exceeded (%d), moving to DLX: %v", retryCount, err)
-			_ = d.Ack(false)
-			return
-		}
-		log.Printf("comment worker: failed (retry %d/%d): %v", retryCount+1, rabbitmq.MaxRetryCount, err)
+	const maxRetries = 3
+	for i := 0; i <= maxRetries; i++ {
+		select {
+		case <-ctx.Done():
 			_ = d.Nack(false, true)
 			return
+		default:
+		}
+		if err := w.process(ctx, d.Body); err != nil {
+			if i >= maxRetries {
+				log.Printf("comment worker: 重试 %d 次后仍失败, 丢弃: %v", maxRetries, err)
+				_ = d.Ack(false)
+				return
+			}
+			wait := time.Duration(1<<uint(i)) * time.Second
+			log.Printf("comment worker: 处理失败, %v 后重试 (%d/%d): %v", wait, i+1, maxRetries, err)
+			time.Sleep(wait)
+			continue
 		}
 		_ = d.Ack(false)
+		return
+	}
 }
 
 func (w *CommentWorker) process(ctx context.Context, body []byte) error {

backend/internal/worker/likeworker.go

@@ -58,18 +58,28 @@ func (w *LikeWorker) Run(ctx context.Context) error {
 }
 
 func (w *LikeWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
-	if err := w.process(ctx, d.Body); err != nil {
-		retryCount := rabbitmq.GetRetryCount(d)
-		if retryCount >= rabbitmq.MaxRetryCount {
-			log.Printf("like worker: max retries exceeded (%d), moving to DLX: %v", retryCount, err)
-			_ = d.Ack(false)
-			return
-		}
-		log.Printf("like worker: failed (retry %d/%d): %v", retryCount+1, rabbitmq.MaxRetryCount, err)
+	const maxRetries = 3
+	for i := 0; i <= maxRetries; i++ {
+		select {
+		case <-ctx.Done():
 			_ = d.Nack(false, true)
 			return
+		default:
+		}
+		if err := w.process(ctx, d.Body); err != nil {
+			if i >= maxRetries {
+				log.Printf("like worker: 重试 %d 次后仍失败, 丢弃: %v", maxRetries, err)
+				_ = d.Ack(false)
+				return
+			}
+			wait := time.Duration(1<<uint(i)) * time.Second
+			log.Printf("like worker: 处理失败, %v 后重试 (%d/%d): %v", wait, i+1, maxRetries, err)
+			time.Sleep(wait)
+			continue
 		}
 		_ = d.Ack(false)
+		return
+	}
 }
 
 func (w *LikeWorker) process(ctx context.Context, body []byte) error {

backend/internal/worker/notificationworker.go

@@ -66,18 +66,28 @@ func (w *NotificationWorker) Run(ctx context.Context) error {
 }
 
 func (w *NotificationWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
-	retryCount := rabbitmq.GetRetryCount(d)
-	if err := w.process(ctx, d); err != nil {
-		if retryCount >= rabbitmq.MaxRetryCount {
-			log.Printf("notification worker: max retries, dropping: %v", err)
-			_ = d.Ack(false)
-			return
-		}
-		log.Printf("notification worker: failed (retry %d/%d): %v", retryCount+1, rabbitmq.MaxRetryCount, err)
+	const maxRetries = 3
+	for i := 0; i <= maxRetries; i++ {
+		select {
+		case <-ctx.Done():
 			_ = d.Nack(false, true)
 			return
+		default:
+		}
+		if err := w.process(ctx, d); err != nil {
+			if i >= maxRetries {
+				log.Printf("notification worker: 重试 %d 次后仍失败, 丢弃: %v", maxRetries, err)
+				_ = d.Ack(false)
+				return
+			}
+			wait := time.Duration(1<<uint(i)) * time.Second
+			log.Printf("notification worker: 处理失败, %v 后重试 (%d/%d): %v", wait, i+1, maxRetries, err)
+			time.Sleep(wait)
+			continue
 		}
 		_ = d.Ack(false)
+		return
+	}
 }
 
 func (w *NotificationWorker) process(ctx context.Context, d amqp.Delivery) error {

backend/internal/worker/popularityworker.go

@@ -8,6 +8,7 @@ import (
 	rediscache "feedsystem_video_go/internal/middleware/redis"
 	"feedsystem_video_go/internal/video"
 	"log"
+	"time"
 
 	amqp "github.com/rabbitmq/amqp091-go"
 )
@@ -57,18 +58,28 @@ func (w *PopularityWorker) Run(ctx context.Context) error {
 }
 
 func (w *PopularityWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
-	if err := w.process(ctx, d.Body); err != nil {
-		retryCount := rabbitmq.GetRetryCount(d)
-		if retryCount >= rabbitmq.MaxRetryCount {
-			log.Printf("popularity worker: max retries exceeded (%d), moving to DLX: %v", retryCount, err)
-			_ = d.Ack(false)
-			return
-		}
-		log.Printf("popularity worker: failed (retry %d/%d): %v", retryCount+1, rabbitmq.MaxRetryCount, err)
+	const maxRetries = 3
+	for i := 0; i <= maxRetries; i++ {
+		select {
+		case <-ctx.Done():
 			_ = d.Nack(false, true)
 			return
+		default:
+		}
+		if err := w.process(ctx, d.Body); err != nil {
+			if i >= maxRetries {
+				log.Printf("popularity worker: 重试 %d 次后仍失败, 丢弃: %v", maxRetries, err)
+				_ = d.Ack(false)
+				return
+			}
+			wait := time.Duration(1<<uint(i)) * time.Second
+			log.Printf("popularity worker: 处理失败, %v 后重试 (%d/%d): %v", wait, i+1, maxRetries, err)
+			time.Sleep(wait)
+			continue
 		}
 		_ = d.Ack(false)
+		return
+	}
 }
 
 func (w *PopularityWorker) process(ctx context.Context, body []byte) error {

backend/internal/worker/socialworker.go

@@ -7,6 +7,7 @@ import (
 	"feedsystem_video_go/internal/middleware/rabbitmq"
 	"feedsystem_video_go/internal/social"
 	"log"
+	"time"
 
 	"github.com/go-sql-driver/mysql"
 	amqp "github.com/rabbitmq/amqp091-go"
@@ -57,18 +58,28 @@ func (w *SocialWorker) Run(ctx context.Context) error {
 }
 
 func (w *SocialWorker) handleDelivery(ctx context.Context, d amqp.Delivery) {
-	if err := w.process(ctx, d.Body); err != nil {
-		retryCount := rabbitmq.GetRetryCount(d)
-		if retryCount >= rabbitmq.MaxRetryCount {
-			log.Printf("social worker: max retries exceeded (%d), moving to DLX: %v", retryCount, err)
-			_ = d.Ack(false)
-			return
-		}
-		log.Printf("social worker: failed (retry %d/%d): %v", retryCount+1, rabbitmq.MaxRetryCount, err)
+	const maxRetries = 3
+	for i := 0; i <= maxRetries; i++ {
+		select {
+		case <-ctx.Done():
 			_ = d.Nack(false, true)
 			return
+		default:
+		}
+		if err := w.process(ctx, d.Body); err != nil {
+			if i >= maxRetries {
+				log.Printf("social worker: 重试 %d 次后仍失败, 丢弃: %v", maxRetries, err)
+				_ = d.Ack(false)
+				return
+			}
+			wait := time.Duration(1<<uint(i)) * time.Second
+			log.Printf("social worker: 处理失败, %v 后重试 (%d/%d): %v", wait, i+1, maxRetries, err)
+			time.Sleep(wait)
+			continue
 		}
 		_ = d.Ack(false)
+		return
+	}
 }
 
 func (w *SocialWorker) process(ctx context.Context, body []byte) error {