feat: 为用到缓存的部分套上了防击穿锁

2025-12-24 14:09:10 +08:00
parent 0d34f2abee
commit a7c6525f57
2 changed files with 150 additions and 36 deletions
--- 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)
--- 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
+						}
+					}
+				}
+			}
 		}
 	}