From 9eed15efd5213fb1f088115f81eef6f7dc8e4754 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Sat, 14 Mar 2026 18:57:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=E5=A4=84=E7=90=86=E4=BA=86=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E6=83=85=E5=86=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/feed/service.go | 26 ++++++++++--------- backend/internal/http/router.go | 4 +++ .../middleware/rabbitmq/timelineMQ.go | 2 +- backend/internal/middleware/redis/cache.go | 6 ++--- backend/internal/worker/outboxworker.go | 3 ++- 5 files changed, 23 insertions(+), 18 deletions(-) diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index e6da21b..24aac3c 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -26,7 +26,7 @@ type FeedService struct { } type CachedFeedData struct { - PublicVideos []video.Video `json:"pubilc_videos"` + PublicVideos []video.Video `json:"public_videos"` } func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, rediscache *rediscache.Client) *FeedService { @@ -70,7 +70,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi } cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - results, err := f.rediscache.MGet(cacheCtx, cacheKeys...).Result() + results, err := f.rediscache.MGet(cacheCtx, cacheKeys...) cancel() if err == nil { @@ -105,7 +105,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi //L3:MySQL var wg sync.WaitGroup var mu sync.Mutex - for _, id := range videoIDs { + for _, id := range missedL2 { wg.Add(1) go func(videoID uint) { defer wg.Done() @@ -114,7 +114,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { videoList, err := f.repo.GetByIDs(ctx, []uint{videoID}) - if err != nil { + if err != nil || len(videoList) == 0 { return nil, err } @@ -147,9 +147,13 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi // 查询最新视频 (冷热分离 + 游标分页) 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) + + if err != nil { + return ListLatestResponse{}, err + } + isZsetEmpty := len(zsetTail) == 0 if isZsetEmpty { @@ -269,11 +273,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti } var hasMore bool - if reqTime <= watermark { - hasMore = len(baseVideos) == limit - } else { - hasMore = true - } + hasMore = len(baseVideos) == limit feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID) if err != nil { @@ -537,9 +537,11 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video } func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { - var res []*video.Video + res := make([]*video.Video, 0, len(orderedIDs)) for _, id := range orderedIDs { - res = append(res, dataMap[id]) + if v, exit := dataMap[id]; exit && v != nil { + res = append(res, v) + } } return res } diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index 7f65f84..c0a767d 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -130,6 +130,10 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g } //worker timelineMQ, err := rabbitmq.NewTimelineMQ(rmq) + if err != nil { + log.Printf("timelineMQ init failed (mq disabled): %v", err) + socialMQ = nil + } worker.StartOutboxPoller(db, timelineMQ) worker.StartConsumer(timelineMQ, "video.timeline.update.queue", cache) return r diff --git a/backend/internal/middleware/rabbitmq/timelineMQ.go b/backend/internal/middleware/rabbitmq/timelineMQ.go index f3d1fdc..6060623 100644 --- a/backend/internal/middleware/rabbitmq/timelineMQ.go +++ b/backend/internal/middleware/rabbitmq/timelineMQ.go @@ -36,7 +36,7 @@ func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) { func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error { if t == nil || t.RabbitMQ == nil { - return errors.New("like mq is not initialized") + return errors.New("timelike mq is not initialized") } if videoID == 0 { return errors.New("videoID are required") diff --git a/backend/internal/middleware/redis/cache.go b/backend/internal/middleware/redis/cache.go index d8bae7b..270cda3 100644 --- a/backend/internal/middleware/redis/cache.go +++ b/backend/internal/middleware/redis/cache.go @@ -3,8 +3,6 @@ package redis import ( "context" "time" - - "github.com/redis/go-redis/v9" ) func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, error) { @@ -19,6 +17,6 @@ func (c *Client) Del(ctx context.Context, key string) error { return c.rdb.Del(ctx, key).Err() } -func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) *redis.SliceCmd { - return c.rdb.MGet(cacheCtx, cacheKeys...) +func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) ([]interface{}, error) { + return c.rdb.MGet(cacheCtx, cacheKeys...).Result() } diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index 662b275..ff62317 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -52,6 +52,7 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi if err != nil { log.Printf("注册消费失败") + return } go func() { @@ -69,7 +70,7 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi timelineKey := "feed:global_timeline" err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ Score: float64(event.CreateTime), - Member: fmt.Sprintf("%d", event.ViedoID), + Member: fmt.Sprintf("%d", event.VideoID), }) if err != nil {