diff --git a/internal/feed/service.go b/internal/feed/service.go index 677db74..4408a0f 100644 --- a/internal/feed/service.go +++ b/internal/feed/service.go @@ -20,7 +20,33 @@ func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, cache return &FeedService{repo: repo, likeRepo: likeRepo, cache: cache, cacheTTL: 5 * time.Second} } +// 查询最新视频 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 + } + // 先从缓存中查询 var cacheKey string if viewerAccountID == 0 && f.cache != nil { before := int64(0) @@ -38,30 +64,46 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti 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 + } + } + } + } } } - - videos, err := f.repo.ListLatest(ctx, limit, latestBefore) + // 缓存中没有查询到结果,从数据库中查询 + resp, err := doListLatestFromDB() 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, - } - + // 缓存查询结果 if cacheKey != "" { if b, err := json.Marshal(resp); err == nil { cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) @@ -72,6 +114,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti return resp, nil } +// 按照点赞数查询视频 func (f *FeedService) ListLikesCount(ctx context.Context, limit int, cursor *LikesCountCursor, viewerAccountID uint) (ListLikesCountResponse, error) { videos, err := f.repo.ListLikesCountWithCursor(ctx, limit, cursor) if err != nil { @@ -96,7 +139,31 @@ func (f *FeedService) ListLikesCount(ctx context.Context, limit int, cursor *Lik return resp, nil } +// 按照关注列表查询视频 func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListByFollowingResponse, error) { + doListByFollowingFromDB := func() (ListByFollowingResponse, error) { + videos, err := f.repo.ListByFollowing(ctx, limit, viewerAccountID, latestBefore) + if err != nil { + return ListByFollowingResponse{}, 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 ListByFollowingResponse{}, err + } + resp := ListByFollowingResponse{ + VideoList: feedVideos, + NextTime: nextTime, + HasMore: hasMore, + } + return resp, nil + } var cacheKey string if viewerAccountID != 0 && f.cache != nil { before := int64(0) @@ -113,29 +180,45 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo 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 ListByFollowingResponse + if err := json.Unmarshal(b, &cached); err == nil { + return cached, nil + } + } else { // 缓存未命中,从数据库中查询 + resp, err := doListByFollowingFromDB() + if err != nil { + return ListByFollowingResponse{}, err + } + if b, err := json.Marshal(resp); err == nil { + _ = f.cache.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 { + var cached ListByFollowingResponse + if err := json.Unmarshal(b, &cached); err == nil { + return cached, nil + } + } + } + } } } - videos, err := f.repo.ListByFollowing(ctx, limit, viewerAccountID, latestBefore) + resp, err := doListByFollowingFromDB() if err != nil { return ListByFollowingResponse{}, 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 ListByFollowingResponse{}, err - } - resp := ListByFollowingResponse{ - VideoList: feedVideos, - NextTime: nextTime, - HasMore: hasMore, - } if cacheKey != "" { if b, err := json.Marshal(resp); err == nil { cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) diff --git a/internal/video/video_service.go b/internal/video/video_service.go index 7399eaa..6957c51 100644 --- a/internal/video/video_service.go +++ b/internal/video/video_service.go @@ -69,6 +69,37 @@ func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) if err := json.Unmarshal(b, &cached); err == nil { return &cached, nil } + } else if rediscache.IsMiss(err) { + lockKey := "lock:" + cacheKey + token, locked, _ := vs.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond) + if locked { + defer func() { _ = vs.cache.Unlock(context.Background(), lockKey, token) }() + if b, err := vs.cache.GetBytes(cacheCtx, cacheKey); err == nil { + var cached Video + if err := json.Unmarshal(b, &cached); err == nil { + return &cached, nil + } + } else { // 缓存未命中,从数据库中查询 + video, err := vs.repo.GetByID(ctx, id) + if err != nil { + return nil, err + } + if b, err := json.Marshal(video); err == nil { + _ = vs.cache.SetBytes(cacheCtx, cacheKey, b, vs.cacheTTL) + } + return video, nil + } + } else { // 缓存未命中,其他goroutine正在查询,等待 + for i := 0; i < 5; i++ { + time.Sleep(20 * time.Millisecond) + if b, err := vs.cache.GetBytes(cacheCtx, cacheKey); err == nil { + var cached Video + if err := json.Unmarshal(b, &cached); err == nil { + return &cached, nil + } + } + } + } } }