diff --git a/.github/workflows/ci.yml b/.gitea/workflows/ci.yml similarity index 92% rename from .github/workflows/ci.yml rename to .gitea/workflows/ci.yml index 3ed2c6d..5255426 100644 --- a/.github/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -7,11 +7,8 @@ on: - main - master -permissions: - contents: read - concurrency: - group: ci-${{ github.workflow }}-${{ github.ref }} + group: ci-${{ gitea.workflow }}-${{ gitea.ref }} cancel-in-progress: true jobs: diff --git a/backend/cmd/worker/main.go b/backend/cmd/worker/main.go index 6ecdc76..b7bf831 100644 --- a/backend/cmd/worker/main.go +++ b/backend/cmd/worker/main.go @@ -17,8 +17,8 @@ import ( "syscall" "time" - amqp "github.com/rabbitmq/amqp091-go" "github.com/joho/godotenv" + amqp "github.com/rabbitmq/amqp091-go" "gorm.io/gorm" ) @@ -40,6 +40,7 @@ const ( popularityBindingKey = "video.popularity.*" ) +// 带重试机制的基础设施连接函数 func connectWithRetry(name string, maxRetries int, fn func() error) { for i := 0; i < maxRetries; i++ { if err := fn(); err == nil { @@ -55,7 +56,7 @@ func connectWithRetry(name string, maxRetries int, fn func() error) { log.Fatalf("%s: 超过最大重试次数", name) } -// runWorkerWithRetry 为每个 Worker 创建独立 Channel,断开后自动重连 +// runWorkerWithRetry 为每个 Worker 创建独立 Channel 并设置 QoS,断开后自动重连 func runWorkerWithRetry(ctx context.Context, name string, conn *amqp.Connection, fn func(*amqp.Channel) error) { for { select { diff --git a/backend/internal/feed/handler.go b/backend/internal/feed/handler.go index 0c91e1f..11ee5f4 100644 --- a/backend/internal/feed/handler.go +++ b/backend/internal/feed/handler.go @@ -168,6 +168,7 @@ func (f *FeedHandler) ListByPopularity(c *gin.Context) { c.JSON(200, resp) } +// 在返回的 FeedVideoItem 列表中,如果列表为 nil,则返回空切片,避免 JSON 序列化为 null func nonNilFeedVideoItems(items []FeedVideoItem) []FeedVideoItem { if items == nil { return []FeedVideoItem{} diff --git a/backend/internal/feed/repo.go b/backend/internal/feed/repo.go index c54c3a3..9a5f5a6 100644 --- a/backend/internal/feed/repo.go +++ b/backend/internal/feed/repo.go @@ -17,6 +17,7 @@ func NewFeedRepository(db *gorm.DB) *FeedRepository { return &FeedRepository{db: db} } +// 查询最新视频. limit: 查询数量 latestBefore: 查询早于此时间的视频(零值则不限制) func (repo *FeedRepository) ListLatest(ctx context.Context, limit int, latestBefore time.Time) ([]*video.Video, error) { var videos []*video.Video query := repo.db.WithContext(ctx).Model(&video.Video{}). @@ -30,6 +31,7 @@ func (repo *FeedRepository) ListLatest(ctx context.Context, limit int, latestBef return videos, nil } +// 查询点赞数最多的视频. limit: 查询数量 cursor: 游标 func (repo *FeedRepository) ListLikesCountWithCursor(ctx context.Context, limit int, cursor *LikesCountCursor) ([]*video.Video, error) { var videos []*video.Video query := repo.db.WithContext(ctx).Model(&video.Video{}). @@ -49,6 +51,7 @@ func (repo *FeedRepository) ListLikesCountWithCursor(ctx context.Context, limit return videos, nil } +// 查询关注用户的视频. limit: 查询数量 viewerAccountID: 查看者账户ID latestBefore: 查询早于此时间的视频(零值则不限制) func (repo *FeedRepository) ListByFollowing(ctx context.Context, limit int, viewerAccountID uint, latestBefore time.Time) ([]*video.Video, error) { var videos []*video.Video query := repo.db.WithContext(ctx).Model(&video.Video{}). @@ -69,6 +72,7 @@ func (repo *FeedRepository) ListByFollowing(ctx context.Context, limit int, view return videos, nil } +// 查询热门视频. limit: 查询数量 popularityBefore: 查询热度低于此值的视频 timeBefore: 查询早于此时间的视频 idBefore: 查询ID小于此值的视频 func (repo *FeedRepository) ListByPopularity(ctx context.Context, limit int, popularityBefore int64, timeBefore time.Time, idBefore uint) ([]*video.Video, error) { var videos []*video.Video query := repo.db.WithContext(ctx).Model(&video.Video{}). diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index f8ec541..8589344 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -33,9 +33,9 @@ func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, redisc return &FeedService{repo: repo, likeRepo: likeRepo, rediscache: rediscache, localcache: cache.New(3*time.Second, 5*time.Second), cacheTTL: 24 * time.Hour} } +// GetVideoByIDs 批量获取视频信息 +// 采用 L1(本地缓存) -> L2(Redis) -> L3(MySQL) 三级架构 func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*video.Video, error) { - // GetVideoByIDs 批量获取视频信息 - // 采用 L1(本地缓存) -> L2(Redis) -> L3(MySQL) 三级架构 if len(videoIDs) == 0 { return []*video.Video{}, nil } @@ -109,6 +109,8 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi wg.Add(1) go func(videoID uint) { defer wg.Done() + + // singleflight 防止缓存击穿 sfKey := f.rediscache.Key("sf:entity:%d", videoID) v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { @@ -160,6 +162,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti isZsetEmpty := len(zsetTail) == 0 + // ZSet 为空时尝试重建 ZSet if isZsetEmpty { //全局静态锁:无视所有用户的不同时间戳游标 sfKey := f.rediscache.Key("sf:fallback:global_timeline_rebuild") @@ -195,10 +198,11 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti return ListLatestResponse{HasMore: false}, nil } - // 让所有被阻塞的请求重新查一遍 + // 递归调用自己,让所有被阻塞的请求重新查一遍 return f.ListLatest(ctx, limit, latestBefore, viewerAccountID) } + // watermark 是 ZSET 中最老的一条数据的时间戳; reqTime 是本次请求的时间戳(如果没有传 latestBefore,则使用当前时间) watermark := int64(zsetTail[0].Score) reqTime := time.Now().UnixMilli() if !latestBefore.IsZero() { @@ -228,11 +232,19 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复 } - videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, f.rediscache.Key("feed:global_timeline"), maxScore, "-inf", 0, int64(limit)) + videoIDsStr, err := f.rediscache.ZRevRangeByScore( + ctx, + f.rediscache.Key("feed:global_timeline"), + maxScore, + "-inf", + 0, + int64(limit), + ) if err != nil { return ListLatestResponse{}, err } + // 将字符串 ID 转换为 uint var videoIDs []uint for _, idStr := range videoIDsStr { if id, err := strconv.ParseUint(idStr, 10, 64); err == nil { @@ -247,7 +259,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti } } - // 刚好击穿了冷热边界 + // 刚好击穿了冷热边界,从数据库中再拉一些冷数据补齐 if len(baseVideos) < limit { remainLimit := limit - len(baseVideos) // 计算还差几个 @@ -381,6 +393,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo token, locked, _ := f.rediscache.Lock(cacheCtx, lockKey, 500*time.Millisecond) if locked { defer func() { _ = f.rediscache.Unlock(context.Background(), lockKey, token) }() + // Double check:再次检查缓存是否被其他请求回写 if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { @@ -396,7 +409,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo } return resp, nil } - } else { + } else { // 加锁失败,循环等待缓存被其他请求回写 for i := 0; i < 5; i++ { time.Sleep(20 * time.Millisecond) if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { @@ -410,11 +423,12 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo } } + // 缓存未命中或 Redis 不可用,降级成数据库查询 resp, err := doListByFollowingFromDB() if err != nil { return ListByFollowingResponse{}, err } - if cacheKey != "" { + if cacheKey != "" { // 缓存回写 if b, err := json.Marshal(resp); err == nil { cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() @@ -427,22 +441,25 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf int64, offset int, viewerAccountID uint, latestPopularity int64, latestBefore time.Time, latestIDBefore uint) (ListByPopularityResponse, error) { // Redis 热榜(稳定分页:as_of + offset) if f.rediscache != nil { + // 将 as_of 截断到分钟级 asOf := time.Now().UTC().Truncate(time.Minute) if reqAsOf > 0 { asOf = time.Unix(reqAsOf, 0).UTC().Truncate(time.Minute) } + // 创建时间窗口,获取过去 60 分钟的 ZSET const win = 60 keys := make([]string, 0, win) for i := 0; i < win; i++ { keys = append(keys, f.rediscache.Key("hot:video:1m:%s", asOf.Add(-time.Duration(i)*time.Minute).Format("200601021504"))) } - dest := f.rediscache.Key("hot:video:merge:1m:%s", asOf.Format("200601021504")) // 快照key:同一个as_of页内复用 + // 创建快照key:同一个as_of页内复用 + dest := f.rediscache.Key("hot:video:merge:1m:%s", asOf.Format("200601021504")) opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond) defer cancel() - exists, _ := f.rediscache.Exists(opCtx, dest) + exists, _ := f.rediscache.Exists(opCtx, dest) // 检查合并快照是否已存在 if !exists { _ = f.rediscache.ZUnionStore(opCtx, dest, keys, "SUM") _ = f.rediscache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间 @@ -462,6 +479,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i } } if err == nil && len(members) > 0 { + // 将字符串 ID 转换为 uint ids := make([]uint, 0, len(members)) for _, m := range members { u, err := strconv.ParseUint(m, 10, 64) @@ -470,6 +488,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i } } + // 根据 ID 批量获取视频信息 videos, err := f.repo.GetByIDs(ctx, ids) if err == nil { byID := make(map[uint]*video.Video, len(videos)) @@ -477,7 +496,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i byID[v.ID] = v } ordered := make([]*video.Video, 0, len(ids)) - for _, id := range ids { + for _, id := range ids { // 按 Redis 返回的顺序重新排列 if v := byID[id]; v != nil { ordered = append(ordered, v) } @@ -492,7 +511,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i NextOffset: offset + len(items), HasMore: len(items) == limit, } - if len(ordered) > 0 { + if len(ordered) > 0 { // 准备最后一条视频的游标信息,供 DB fallback 使用 last := ordered[len(ordered)-1] nextPopularity := last.Popularity nextBefore := last.CreateTime @@ -506,6 +525,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i } } + // DB fallback(游标分页:latestPopularity + latestBefore + latestIDBefore) videos, err := f.repo.ListByPopularity(ctx, limit, latestPopularity, latestBefore, latestIDBefore) if err != nil { return ListByPopularityResponse{}, err @@ -532,6 +552,7 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i return resp, nil } +// 将 Video 列表转换为 FeedVideoItem 列表,并批量查询填充当前用户对所有视频的点赞状态 func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video, viewerAccountID uint) ([]FeedVideoItem, error) { feedVideos := make([]FeedVideoItem, 0, len(videos)) videoIDs := make([]uint, len(videos)) @@ -558,6 +579,7 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video return feedVideos, nil } +// 将视频列表按照给定的 ID 顺序(orderedIDs)重新排序 func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { res := make([]*video.Video, 0, len(orderedIDs)) for _, id := range orderedIDs { diff --git a/backend/internal/middleware/rabbitmq/dlx.go b/backend/internal/middleware/rabbitmq/dlx.go index 82efe5b..c3b9aaa 100644 --- a/backend/internal/middleware/rabbitmq/dlx.go +++ b/backend/internal/middleware/rabbitmq/dlx.go @@ -23,7 +23,7 @@ func DeclareDLX(ch *amqp.Channel, queueName string) error { } dlxQueue := queueName + ".dlx" _, err := ch.QueueDeclare( - dlxQueue, true, false, false, false, nil, + dlxQueue, true, false, false, false, nil, // 死信队列不设置 DLX ) if err != nil { return err diff --git a/backend/internal/middleware/rabbitmq/likeMQ.go b/backend/internal/middleware/rabbitmq/likeMQ.go index 6bfc65d..62e6328 100644 --- a/backend/internal/middleware/rabbitmq/likeMQ.go +++ b/backend/internal/middleware/rabbitmq/likeMQ.go @@ -17,8 +17,8 @@ const ( likeQueue = "like.events" likeBindingKey = "like.*" - likeLikeRK = "like.like" - likeUnlikeRK = "like.unlike" + likeLikeRK = "like.like" // 点赞路由键 + likeUnlikeRK = "like.unlike" // 取消点赞路由键 ) type LikeEvent struct { diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go index 7a2e96d..f290394 100644 --- a/backend/internal/middleware/rabbitmq/rabbitMQ.go +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -69,12 +69,12 @@ func DeclareTopic(ch *amqp.Channel, exchange string, queue string, bindingKey st } q, err := ch.QueueDeclare( - queue, - true, - false, - false, - false, - amqp.Table{"x-dead-letter-exchange": DLXExchange}, + queue, // 队列名称 + true, // 持久化 + false, // autoDelete 是否在未使用时自动删除 + false, // exclusive 是否排他(允许多个消费者共享) + false, // noWait 是否不等待 broker 确认 + amqp.Table{"x-dead-letter-exchange": DLXExchange}, // 死信交换机 ) if err != nil { return err @@ -84,8 +84,8 @@ func DeclareTopic(ch *amqp.Channel, exchange string, queue string, bindingKey st q.Name, bindingKey, exchange, - false, - nil, + false, // noWait + nil, // args ); err != nil { return err } @@ -107,10 +107,10 @@ func PublishJSON(ctx context.Context, ch *amqp.Channel, exchange string, routing return err } return ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ - ContentType: "application/json", - DeliveryMode: amqp.Persistent, - Timestamp: time.Now(), - Body: b, + ContentType: "application/json", // 消息格式 + DeliveryMode: amqp.Persistent, // 消息持久化到磁盘, 值为 1 则不持久化, 值为 2 则持久化, amqp.Presistent 为 2 + Timestamp: time.Now(), // 消息产生的时间戳 + Body: b, // 实际消息内容(JSON字节) }) } diff --git a/backend/internal/social/service.go b/backend/internal/social/service.go index e7a9b5c..b084644 100644 --- a/backend/internal/social/service.go +++ b/backend/internal/social/service.go @@ -91,6 +91,7 @@ func (s *SocialService) Unfollow(ctx context.Context, social *Social) error { return nil } +// 失效关注列表缓存,确保下次查询时能获取到最新的关注列表 func (s *SocialService) invalidateFollowingFeedCache(ctx context.Context, accountID uint) { if s.cache == nil { return diff --git a/backend/internal/video/like_repo.go b/backend/internal/video/like_repo.go index c8d1a76..62a63d2 100644 --- a/backend/internal/video/like_repo.go +++ b/backend/internal/video/like_repo.go @@ -62,6 +62,7 @@ func (r *LikeRepository) IsLiked(ctx context.Context, videoID, accountID uint) ( return count > 0, nil } +// 查询给定视频 ID 列表中,哪些视频被指定账户点赞过 func (r *LikeRepository) BatchGetLiked(ctx context.Context, videoIDs []uint, accountID uint) (map[uint]bool, error) { likeMap := make(map[uint]bool) if len(videoIDs) == 0 { diff --git a/backend/internal/worker/likeworker.go b/backend/internal/worker/likeworker.go index 445f196..7aeefb8 100644 --- a/backend/internal/worker/likeworker.go +++ b/backend/internal/worker/likeworker.go @@ -32,13 +32,13 @@ func (w *LikeWorker) Run(ctx context.Context) error { } deliveries, err := w.ch.Consume( - w.queue, - "", - false, - false, - false, - false, - nil, + w.queue, // 队列名 + "", // 消费者标签,空代表让 RabbitMQ 自动生成一个唯一的标签 + false, // autoAck = false 采用手动确认模式 + false, // exclusive = false 允许多个消费者同时消费同一个队列 + false, // noLocal = false 允许消费者接收自己发送的消息 + false, // noWait = false 阻塞等待 RabbitMQ 的响应 + nil, // args ) if err != nil { return err diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index aa87472..87f8f26 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -14,6 +14,7 @@ import ( "gorm.io/gorm" ) +// 轮询器,轮询数据库中的 outbox 表,获取待投递的消息,投递到 MQ 中 func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { if db == nil || tmq == nil { log.Printf("Outbox poller disabled: timeline mq is not initialized") @@ -46,6 +47,7 @@ func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { }() } +// 消费者,消费 MQ 中的消息,写入 Redis 的 Zset 中 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") @@ -66,6 +68,8 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi continue } + // 设置当前 Channel 的 QoS(Quality of Service)参数,控制消息的预取数量和大小 + // prefetch count = 10 一次最多取10个消息, prefetch size = 0 不限制预取的字节大小, global = false Qos 设置只对当前 Channel 生效 if err := ch.Qos(10, 0, false); err != nil { log.Printf("Timeline consumer: QoS 设置失败: %v", err) }