From e3af802784b0a96fb43cf369cfca023933e07ee8 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 18:10:46 +0800 Subject: [PATCH] =?UTF-8?q?perf:=E4=BC=98=E5=8C=96=E4=BA=86ListLatest?= =?UTF-8?q?=E5=9C=A8=E5=A4=A7=E8=A7=84=E6=A8=A1=E8=A7=86=E9=A2=91=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=8B=E7=9A=84=E5=93=8D=E5=BA=94=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/go.mod | 1 + backend/go.sum | 2 + backend/internal/feed/service.go | 372 +++++++++++++++++++++++-------- 3 files changed, 279 insertions(+), 96 deletions(-) diff --git a/backend/go.mod b/backend/go.mod index f8d5d70..fd2b43e 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -36,6 +36,7 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/patrickmn/go-cache v2.1.0+incompatible github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/quic-go/qpack v0.5.1 // indirect github.com/quic-go/quic-go v0.54.0 // indirect diff --git a/backend/go.sum b/backend/go.sum index a94cb7d..8063d6f 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -58,6 +58,8 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OH github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= +github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index 6577138..e6da21b 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -6,113 +6,285 @@ import ( rediscache "feedsystem_video_go/internal/middleware/redis" "feedsystem_video_go/internal/video" "fmt" + "log" "strconv" + "sync" "time" + + "github.com/patrickmn/go-cache" + redis "github.com/redis/go-redis/v9" + "golang.org/x/sync/singleflight" ) type FeedService struct { - repo *FeedRepository - likeRepo *video.LikeRepository - cache *rediscache.Client - cacheTTL time.Duration + repo *FeedRepository + likeRepo *video.LikeRepository + rediscache *rediscache.Client + localcache *cache.Cache + cacheTTL time.Duration + requestGroup singleflight.Group } -func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, cache *rediscache.Client) *FeedService { - return &FeedService{repo: repo, likeRepo: likeRepo, cache: cache, cacheTTL: 5 * time.Second} +type CachedFeedData struct { + PublicVideos []video.Video `json:"pubilc_videos"` } -// 查询最新视频 -func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { - // 从数据库中查询最新视频 - doListLatestFromDB := func() (ListLatestResponse, error) { - videos, err := f.repo.ListLatest(ctx, limit, latestBefore) - if err != nil { - return ListLatestResponse{}, err - } - var nextTime int64 - if len(videos) > 0 { - nextTime = videos[len(videos)-1].CreateTime.Unix() - } else { - nextTime = 0 - } - hasMore := len(videos) == limit - feedVideos, err := f.buildFeedVideos(ctx, videos, viewerAccountID) - if err != nil { - return ListLatestResponse{}, err - } - resp := ListLatestResponse{ - VideoList: feedVideos, - NextTime: nextTime, - HasMore: hasMore, - } - return resp, nil +func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, rediscache *rediscache.Client) *FeedService { + return &FeedService{repo: repo, likeRepo: likeRepo, rediscache: rediscache, localcache: cache.New(3*time.Second, 5*time.Second), cacheTTL: 24 * time.Hour} +} + +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 } - // 先从缓存中查询 - var cacheKey string - if viewerAccountID == 0 && f.cache != nil { - before := int64(0) - if !latestBefore.IsZero() { - before = latestBefore.Unix() + + videoMap := make(map[uint]*video.Video) + //L1:本地缓存 + var missedL1 []uint + for _, id := range videoIDs { + cacheKey := fmt.Sprintf("video:entity:%d", id) + if f.localcache != nil { + if v, found := f.localcache.Get(cacheKey); found { + if data, ok := v.(video.Video); ok { + videoMap[id] = &data + continue + } + } + } + // 记录未命中的 ID，准备进入下一级缓存 + missedL1 = append(missedL1, id) + } + + if len(missedL1) == 0 { + return buildOrderedResult(videoIDs, videoMap), nil + } + + //L2:redis + var missedL2 []uint + if len(missedL1) > 0 { + cacheKeys := make([]string, len(missedL1)) + for i, id := range missedL1 { + cacheKeys[i] = fmt.Sprintf("video:entity:%d", id) } - cacheKey = fmt.Sprintf("feed:listLatest:limit=%d:before=%d", limit, before) cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() + results, err := f.rediscache.MGet(cacheCtx, cacheKeys...).Result() + cancel() - b, err := f.cache.GetBytes(cacheCtx, cacheKey) if err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil - } - } else if rediscache.IsMiss(err) { // 缓存未命中 - lockKey := "lock:" + cacheKey - // 缓存未命中，尝试加锁 - token, locked, _ := f.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond) - if locked { - defer func() { _ = f.cache.Unlock(context.Background(), lockKey, token) }() - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil - } - } else { // 缓存未命中，从数据库中查询 - resp, err := doListLatestFromDB() - if err != nil { - return ListLatestResponse{}, err - } - if b, err := json.Marshal(resp); err == nil { - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) - } - return resp, nil - } - } else { // 缓存未命中，其他goroutine正在查询，等待 - for i := 0; i < 5; i++ { - time.Sleep(20 * time.Millisecond) - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil + for i, res := range results { + id := missedL1[i] + if res != nil { + if str, ok := res.(string); ok { + var v video.Video + if err := json.Unmarshal([]byte(str), &v); err == nil { + videoMap[id] = &v + // 回写更新 L1 本地缓存 + if f.localcache != nil { + f.localcache.Set(cacheKeys[i], v, 5*time.Second) + } + continue } } } + missedL2 = append(missedL2, id) + } + } else { + // 如果 Redis 挂了或者超时了，全部降级到 L3 + missedL2 = missedL1 + log.Printf("L2 Redis MGet 失败，全部降级到 MySQL: %v", err) + } + } + + if len(missedL2) == 0 { + return buildOrderedResult(videoIDs, videoMap), nil + } + + //L3:MySQL + var wg sync.WaitGroup + var mu sync.Mutex + for _, id := range videoIDs { + wg.Add(1) + go func(videoID uint) { + defer wg.Done() + sfKey := fmt.Sprintf("sf:entity:%d", videoID) + + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + videoList, err := f.repo.GetByIDs(ctx, []uint{videoID}) + + if err != nil { + return nil, err + } + + safeCopy := *videoList[0] + cachekey := fmt.Sprintf("video:entity:%d", safeCopy.ID) + if b, err := json.Marshal(safeCopy); err == nil { + //异步回写redis + go func(k string, b []byte) { + setCtx, setCancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer setCancel() + + f.rediscache.SetBytes(setCtx, k, b, time.Hour) + }(cachekey, b) + } + return videoList[0], err + }) + + if err == nil && v != nil { + safeCopy := *(v.(*video.Video)) + mu.Lock() + videoMap[id] = &safeCopy + mu.Unlock() + f.localcache.Set(fmt.Sprintf("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second) + } + }(id) + } + wg.Wait() + return buildOrderedResult(videoIDs, videoMap), nil +} + +// 查询最新视频 (冷热分离 + 游标分页) +func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { + + // 获取 ZSET 中最老的一条数据 + zsetTail, err := f.rediscache.ZRangeWithScores(ctx, "feed:global_timeline", 0, 0) + isZsetEmpty := len(zsetTail) == 0 + + if isZsetEmpty { + //全局静态锁：无视所有用户的不同时间戳游标 + sfKey := "sf:fallback:global_timeline_rebuild" + + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + // 无视游标，直接去 MySQL 捞最新的 1000 条 + dbVideos, err := f.repo.ListLatest(ctx, 1000, time.Time{}) + if err != nil { + return nil, err + } + if len(dbVideos) == 0 { + return "EMPTY_DB", nil // 防无限递归 + } + + // 重建 ZSET + bgCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + var zElements []redis.Z + for _, vid := range dbVideos { + zElements = append(zElements, redis.Z{ + Score: float64(vid.CreateTime.UnixMilli()), + Member: fmt.Sprintf("%d", vid.ID), + }) + } + f.rediscache.ZAdd(bgCtx, "feed:global_timeline", zElements...) + return "SUCCESS", nil + }) + + if err != nil { + return ListLatestResponse{}, err + } + if v == "EMPTY_DB" { + return ListLatestResponse{HasMore: false}, nil + } + + // 让所有被阻塞的请求重新查一遍 + return f.ListLatest(ctx, limit, latestBefore, viewerAccountID) + } + + watermark := int64(zsetTail[0].Score) + reqTime := time.Now().UnixMilli() + if !latestBefore.IsZero() { + reqTime = latestBefore.UnixMilli() + } + + var baseVideos []*video.Video + + if reqTime <= watermark { + //冷数据降级查库 + + // 针对个别用户的防并发（此时可以用时间戳做锁，因为冷尾流量极小） + sfKey := fmt.Sprintf("sf:cold:listLatest:%d:%d", limit, reqTime) + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + return f.repo.ListLatest(ctx, limit, latestBefore) + }) + if err != nil { + return ListLatestResponse{}, err + } + baseVideos = v.([]*video.Video) + // 不回写 ZSET，防止冷数据污染热点时间线 + + } else { + // 热数据直接查redis + maxScore := "+inf" + if !latestBefore.IsZero() { + maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复 + } + + videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, "feed:global_timeline", maxScore, "-inf", 0, int64(limit)) + if err != nil { + return ListLatestResponse{}, err + } + + var videoIDs []uint + for _, idStr := range videoIDsStr { + if id, err := strconv.ParseUint(idStr, 10, 64); err == nil { + videoIDs = append(videoIDs, uint(id)) + } + } + + if len(videoIDs) > 0 { + baseVideos, err = f.GetVideoByIDs(ctx, videoIDs) + if err != nil { + return ListLatestResponse{}, err + } + } + + // 刚好击穿了冷热边界 + if len(baseVideos) < limit { + remainLimit := limit - len(baseVideos) // 计算还差几个 + + var coldCursor time.Time + if len(baseVideos) > 0 { + coldCursor = baseVideos[len(baseVideos)-1].CreateTime + } else { + coldCursor = latestBefore + } + + sfKey := fmt.Sprintf("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli()) + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + return f.repo.ListLatest(ctx, remainLimit, coldCursor) + }) + + if err == nil { + coldVideos := v.([]*video.Video) + baseVideos = append(baseVideos, coldVideos...) } } } - // 缓存中没有查询到结果，从数据库中查询 - resp, err := doListLatestFromDB() + + var nextTime int64 + if len(baseVideos) > 0 { + // 将本页最后一条视频的时间作为下一次请求的游标 + nextTime = baseVideos[len(baseVideos)-1].CreateTime.UnixMilli() + } + var hasMore bool + + if reqTime <= watermark { + hasMore = len(baseVideos) == limit + } else { + hasMore = true + } + + feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID) if err != nil { return ListLatestResponse{}, err } - // 缓存查询结果 - if cacheKey != "" { - if b, err := json.Marshal(resp); err == nil { - cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) - } - } - return resp, nil + + return ListLatestResponse{ + VideoList: feedVideos, + NextTime: nextTime, + HasMore: hasMore, + }, nil } // 按照点赞数查询视频 @@ -166,7 +338,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo return resp, nil } var cacheKey string - if viewerAccountID != 0 && f.cache != nil { + if viewerAccountID != 0 && f.rediscache != nil { before := int64(0) if !latestBefore.IsZero() { before = latestBefore.Unix() @@ -175,7 +347,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() - b, err := f.cache.GetBytes(cacheCtx, cacheKey) + b, err := f.rediscache.GetBytes(cacheCtx, cacheKey) if err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { @@ -184,10 +356,10 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo } else if rediscache.IsMiss(err) { // 缓存未命中 lockKey := "lock:" + cacheKey // 缓存未命中，尝试加锁 - token, locked, _ := f.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond) + token, locked, _ := f.rediscache.Lock(cacheCtx, lockKey, 500*time.Millisecond) if locked { - defer func() { _ = f.cache.Unlock(context.Background(), lockKey, token) }() - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { + defer func() { _ = f.rediscache.Unlock(context.Background(), lockKey, token) }() + if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { return cached, nil @@ -198,14 +370,14 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo return ListByFollowingResponse{}, err } if b, err := json.Marshal(resp); err == nil { - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) + _ = f.rediscache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) } return resp, nil } } else { for i := 0; i < 5; i++ { time.Sleep(20 * time.Millisecond) - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { + if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { return cached, nil @@ -224,7 +396,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo if b, err := json.Marshal(resp); err == nil { cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) + _ = f.rediscache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) } } return resp, nil @@ -232,7 +404,7 @@ 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.cache != nil { + if f.rediscache != nil { asOf := time.Now().UTC().Truncate(time.Minute) if reqAsOf > 0 { asOf = time.Unix(reqAsOf, 0).UTC().Truncate(time.Minute) @@ -248,15 +420,15 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond) defer cancel() - exists, _ := f.cache.Exists(opCtx, dest) + exists, _ := f.rediscache.Exists(opCtx, dest) if !exists { - _ = f.cache.ZUnionStore(opCtx, dest, keys, "SUM") - _ = f.cache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间 + _ = f.rediscache.ZUnionStore(opCtx, dest, keys, "SUM") + _ = f.rediscache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间 } start := int64(offset) stop := start + int64(limit) - 1 - members, err := f.cache.ZRevRange(opCtx, dest, start, stop) + members, err := f.rediscache.ZRevRange(opCtx, dest, start, stop) if err == nil && len(members) == 0 { if offset > 0 { return ListByPopularityResponse{ @@ -363,3 +535,11 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video } return feedVideos, nil } + +func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { + var res []*video.Video + for _, id := range orderedIDs { + res = append(res, dataMap[id]) + } + return res +}