Merge pull request #1 from Amnesia-debug/feat/feed-architecture-upgrade

Feat/feed architecture upgrade

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

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=

backend/internal/db/db.go

@@ -24,7 +24,7 @@ func NewDB(dbcfg config.DatabaseConfig) (*gorm.DB, error) {
 }
 
 func AutoMigrate(db *gorm.DB) error {
-	return db.AutoMigrate(&account.Account{}, &video.Video{}, &video.Like{}, &video.Comment{}, &social.Social{})
+	return db.AutoMigrate(&account.Account{}, &video.Video{}, &video.Like{}, &video.Comment{}, &social.Social{}, &video.OutboxMsg{})
 }
 
 func CloseDB(db *gorm.DB) error {

backend/internal/feed/handler.go

@@ -26,7 +26,7 @@ func (f *FeedHandler) ListLatest(c *gin.Context) {
 	}
 	var latestTime time.Time
 	if req.LatestTime > 0 {
-		latestTime = time.Unix(req.LatestTime, 0)
+		latestTime = time.UnixMilli(req.LatestTime)
 	}
 	viewerAccountID, err := jwt.GetAccountID(c)
 	if err != nil {

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
+	repo         *FeedRepository
-	likeRepo *video.LikeRepository
+	likeRepo     *video.LikeRepository
-	cache    *rediscache.Client
+	rediscache   *rediscache.Client
-	cacheTTL time.Duration
+	localcache   *cache.Cache
+	cacheTTL     time.Duration
+	requestGroup singleflight.Group
 }
 
-func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, cache *rediscache.Client) *FeedService {
+type CachedFeedData struct {
-	return &FeedService{repo: repo, likeRepo: likeRepo, cache: cache, cacheTTL: 5 * time.Second}
+	PublicVideos []video.Video `json:"public_videos"`
 }
 
-// 查询最新视频
+func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, rediscache *rediscache.Client) *FeedService {
-func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) {
+	return &FeedService{repo: repo, likeRepo: likeRepo, rediscache: rediscache, localcache: cache.New(3*time.Second, 5*time.Second), cacheTTL: 24 * time.Hour}
-	// 从数据库中查询最新视频
+}
-	doListLatestFromDB := func() (ListLatestResponse, error) {
+
-		videos, err := f.repo.ListLatest(ctx, limit, latestBefore)
+func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*video.Video, error) {
-		if err != nil {
+	// GetVideoByIDs 批量获取视频信息
-			return ListLatestResponse{}, err
+	// 采用 L1(本地缓存) -> L2(Redis) -> L3(MySQL) 三级架构
-		}
+	if len(videoIDs) == 0 {
-		var nextTime int64
+		return []*video.Video{}, nil
-		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
+	videoMap := make(map[uint]*video.Video)
-	if viewerAccountID == 0 && f.cache != nil {
+	//L1:本地缓存
-		before := int64(0)
+	var missedL1 []uint
-		if !latestBefore.IsZero() {
+	for _, id := range videoIDs {
-			before = latestBefore.Unix()
+		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...)
+		cancel()
 
-		b, err := f.cache.GetBytes(cacheCtx, cacheKey)
 		if err == nil {
-			var cached ListLatestResponse
+			for i, res := range results {
-			if err := json.Unmarshal(b, &cached); err == nil {
+				id := missedL1[i]
-				return cached, nil
+				if res != nil {
-			}
+					if str, ok := res.(string); ok {
-		} else if rediscache.IsMiss(err) { // 缓存未命中
+						var v video.Video
-			lockKey := "lock:" + cacheKey
+						if err := json.Unmarshal([]byte(str), &v); err == nil {
-			// 缓存未命中，尝试加锁
+							videoMap[id] = &v
-			token, locked, _ := f.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond)
+							// 回写更新 L1 本地缓存
-			if locked {
+							if f.localcache != nil {
-				defer func() { _ = f.cache.Unlock(context.Background(), lockKey, token) }()
+								f.localcache.Set(cacheKeys[i], v, 5*time.Second)
-				if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil {
+							}
-					var cached ListLatestResponse
+							continue
-					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
 						}
 					}
 				}
+				missedL2 = append(missedL2, id)
 			}
+		} else {
+			// 如果 Redis 挂了或者超时了，全部降级到 L3
+			missedL2 = missedL1
+			log.Printf("L2 Redis MGet 失败，全部降级到 MySQL: %v", err)
 		}
 	}
-	// 缓存中没有查询到结果，从数据库中查询
+
-	resp, err := doListLatestFromDB()
+	if len(missedL2) == 0 {
+		return buildOrderedResult(videoIDs, videoMap), nil
+	}
+
+	//L3:MySQL
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	for _, id := range missedL2 {
+		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 || len(videoList) == 0 {
+					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)
+
 	if err != nil {
 		return ListLatestResponse{}, err
 	}
-	// 缓存查询结果
+
-	if cacheKey != "" {
+	isZsetEmpty := len(zsetTail) == 0
-		if b, err := json.Marshal(resp); err == nil {
+
-			cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
+	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()
-			_ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL)
+			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...)
+			}
 		}
 	}
-	return resp, nil
+
+	var nextTime int64
+	if len(baseVideos) > 0 {
+		// 将本页最后一条视频的时间作为下一次请求的游标
+		nextTime = baseVideos[len(baseVideos)-1].CreateTime.UnixMilli()
+	}
+	var hasMore bool
+
+	hasMore = len(baseVideos) == limit
+
+	feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID)
+	if err != nil {
+		return ListLatestResponse{}, err
+	}
+
+	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) }()
+				defer func() { _ = f.rediscache.Unlock(context.Background(), lockKey, token) }()
-				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
@@ -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.rediscache.ZUnionStore(opCtx, dest, keys, "SUM")
-			_ = f.cache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间
+			_ = 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,13 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video
 	}
 	return feedVideos, nil
 }
+
+func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video {
+	res := make([]*video.Video, 0, len(orderedIDs))
+	for _, id := range orderedIDs {
+		if v, exits := dataMap[id]; exits && v != nil {
+			res = append(res, v)
+		}
+	}
+	return res
+}

backend/internal/http/router.go

@@ -8,6 +8,7 @@ import (
 	rediscache "feedsystem_video_go/internal/middleware/redis"
 	"feedsystem_video_go/internal/social"
 	"feedsystem_video_go/internal/video"
+	"feedsystem_video_go/internal/worker"
 	"log"
 
 	"github.com/gin-gonic/gin"
@@ -127,5 +128,13 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g
 	{
 		protectedFeedGroup.POST("/listByFollowing", feedHandler.ListByFollowing)
 	}
+	//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
 }

backend/internal/middleware/rabbitmq/rabbitMQ.go

@@ -14,8 +14,8 @@ import (
 )
 
 type RabbitMQ struct {
-	conn *amqp.Connection
+	Conn *amqp.Connection
-	ch   *amqp.Channel
+	Ch   *amqp.Channel
 }
 
 func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) {
@@ -31,31 +31,31 @@ func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &RabbitMQ{conn: conn, ch: ch}, nil
+	return &RabbitMQ{Conn: conn, Ch: ch}, nil
 }
 
 func (r *RabbitMQ) Close() error {
-	if r == nil || r.ch == nil || r.conn == nil {
+	if r == nil || r.Ch == nil || r.Conn == nil {
 		return nil
 	}
-	if err := r.ch.Close(); err != nil {
+	if err := r.Ch.Close(); err != nil {
 		return err
 	}
-	if err := r.conn.Close(); err != nil {
+	if err := r.Conn.Close(); err != nil {
 		return err
 	}
 	return nil
 }
 
 func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error {
-	if r == nil || r.ch == nil {
+	if r == nil || r.Ch == nil {
 		return errors.New("rabbitmq is not initialized")
 	}
 	if exchange == "" || queue == "" || bindingKey == "" {
 		return errors.New("exchange/queue/bindingKey is required")
 	}
 
-	if err := r.ch.ExchangeDeclare(
+	if err := r.Ch.ExchangeDeclare(
 		exchange,
 		"topic",
 		true,
@@ -67,7 +67,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string
 		return err
 	}
 
-	q, err := r.ch.QueueDeclare(
+	q, err := r.Ch.QueueDeclare(
 		queue,
 		true,
 		false,
@@ -79,7 +79,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string
 		return err
 	}
 
-	return r.ch.QueueBind(
+	return r.Ch.QueueBind(
 		q.Name,
 		bindingKey,
 		exchange,
@@ -89,7 +89,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string
 }
 
 func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error {
-	if r == nil || r.ch == nil {
+	if r == nil || r.Ch == nil {
 		return errors.New("rabbitmq is not initialized")
 	}
 	if exchange == "" || routingKey == "" {
@@ -99,7 +99,7 @@ func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey
 	if err != nil {
 		return err
 	}
-	return r.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
+	return r.Ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
 		ContentType:  "application/json",
 		DeliveryMode: amqp.Persistent,
 		Timestamp:    time.Now(),

backend/internal/middleware/rabbitmq/timelineMQ.go

@@ -0,0 +1,55 @@
+package rabbitmq
+
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+type TimelineMQ struct {
+	*RabbitMQ
+}
+
+const (
+	timelineExchange   = "video.timeline.events"
+	timelineQueue      = "video.timeline.update.queue"
+	timelineBindingKey = "video.timeline.*"
+	timelinePublishRK  = "video.timeline.publish"
+)
+
+type TimelineEvent struct {
+	EventID    string    `json:"event_id"`
+	VideoID    uint      `json:"video_id"`
+	CreateTime int64     `json:"create_time"`
+	OccurredAt time.Time `json:"occurred_at"`
+}
+
+func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) {
+	if base == nil {
+		return nil, errors.New("rabbitmq base is nil")
+	}
+	if err := base.DeclareTopic(timelineExchange, timelineQueue, timelineBindingKey); err != nil {
+		return nil, err
+	}
+	return &TimelineMQ{RabbitMQ: base}, nil
+}
+
+func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error {
+	if t == nil || t.RabbitMQ == nil {
+		return errors.New("timeline mq is not initialized")
+	}
+	if videoID == 0 {
+		return errors.New("videoID are required")
+	}
+	id, err := newEventID(16)
+	if err != nil {
+		return err
+	}
+	timeline := TimelineEvent{
+		EventID:    id,
+		VideoID:    videoID,
+		CreateTime: createTime.UnixMilli(),
+		OccurredAt: time.Now(),
+	}
+	return t.PublishJSON(ctx, timelineExchange, timelinePublishRK, timeline)
+}

backend/internal/middleware/redis/cache.go

@@ -16,3 +16,7 @@ func (c *Client) SetBytes(ctx context.Context, key string, value []byte, ttl tim
 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) ([]interface{}, error) {
+	return c.rdb.MGet(cacheCtx, cacheKeys...).Result()
+}

backend/internal/middleware/redis/zset.go

@@ -2,6 +2,7 @@ package redis
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	redis "github.com/redis/go-redis/v9"
@@ -14,6 +15,27 @@ func (c *Client) ZincrBy(ctx context.Context, key string, member string, score f
 	return c.rdb.ZIncrBy(ctx, key, score, member).Err()
 }
 
+func (c *Client) ZAdd(ctx context.Context, key string, members ...redis.Z) error {
+	if c == nil || c.rdb == nil {
+		return nil
+	}
+	return c.rdb.ZAdd(ctx, key, members...).Err()
+}
+
+func (c *Client) ZRemRangeByRank(ctx context.Context, key string, start int64, stop int64) error {
+	if c == nil || c.rdb == nil {
+		return nil
+	}
+	return c.rdb.ZRemRangeByRank(ctx, key, start, stop).Err()
+}
+
+func (c *Client) ZRangeWithScores(ctx context.Context, key string, start int64, stop int64) ([]redis.Z, error) {
+	if c == nil || c.rdb == nil {
+		return nil, errors.New("redis client not initialized")
+	}
+	return c.rdb.ZRangeWithScores(ctx, key, start, stop).Result()
+}
+
 func (c *Client) Expire(ctx context.Context, key string, ttl time.Duration) error {
 	if c == nil || c.rdb == nil {
 		return nil

backend/internal/video/video_entity.go

@@ -38,3 +38,11 @@ type UpdateLikesCountRequest struct {
 	ID         uint  `json:"id"`
 	LikesCount int64 `json:"likes_count"`
 }
+
+type OutboxMsg struct {
+	ID         uint      `gorm:"primaryKey"`
+	VideoID    uint      `gorm:"index"`
+	EventType  string    `gorm:"type:varchar(50)"`
+	CreateTime time.Time `gorm:"autoCreateTime"`
+	Status     string    `gorm:"type:varchar(50);index"`
+}

backend/internal/video/video_repo.go

@@ -22,6 +22,13 @@ func (vr *VideoRepository) CreateVideo(ctx context.Context, video *Video) error
 	return nil
 }
 
+func (vr *VideoRepository) CreateMsg(ctx context.Context, Msg *OutboxMsg) error {
+	if err := vr.db.WithContext(ctx).Create(Msg).Error; err != nil {
+		return err
+	}
+	return nil
+}
+
 func (vr *VideoRepository) DeleteVideo(ctx context.Context, id uint) error {
 	if err := vr.db.WithContext(ctx).Delete(&Video{}, id).Error; err != nil {
 		return err

backend/internal/video/video_service.go

@@ -11,6 +11,8 @@ import (
 
 	"feedsystem_video_go/internal/middleware/rabbitmq"
 	rediscache "feedsystem_video_go/internal/middleware/redis"
+
+	"gorm.io/gorm"
 )
 
 type VideoService struct {
@@ -41,10 +43,28 @@ func (vs *VideoService) Publish(ctx context.Context, video *Video) error {
 	if video.CoverURL == "" {
 		return errors.New("cover url is required")
 	}
-	if err := vs.repo.CreateVideo(ctx, video); err != nil {
+
-		return err
+	//事务保证视频写入库和消息写入本地消息表的一致性
-	}
+	err := vs.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-	return nil
+		if err := tx.Create(video).Error; err != nil {
+			return err
+		}
+
+		msg := OutboxMsg{
+			VideoID:    video.ID,
+			EventType:  "video_published",
+			Status:     "pending",
+			CreateTime: video.CreateTime,
+		}
+
+		if err := tx.Create(&msg).Error; err != nil {
+			return err
+		}
+		return nil
+
+	})
+	return err
+
 }
 
 func (vs *VideoService) Delete(ctx context.Context, id uint, authorID uint) error {

backend/internal/worker/outboxworker.go

@@ -0,0 +1,93 @@
+package worker
+
+import (
+	"context"
+	"encoding/json"
+	"feedsystem_video_go/internal/middleware/rabbitmq"
+	"feedsystem_video_go/internal/middleware/redis"
+	"feedsystem_video_go/internal/video"
+	"fmt"
+	"log"
+	"time"
+
+	oredis "github.com/redis/go-redis/v9"
+	"gorm.io/gorm"
+)
+
+func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) {
+	go func() {
+		for {
+			var messages []video.OutboxMsg
+
+			err := db.Where("status = ?", "pending").Order("create_time ASC").Limit(100).Find(&messages).Error
+
+			if err != nil || len(messages) == 0 {
+				time.Sleep(1 * time.Second)
+				continue
+			}
+
+			for _, msg := range messages {
+				err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime)
+
+				if err == nil {
+					db.Delete(&msg)
+				} else {
+					log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err)
+				}
+			}
+		}
+	}()
+}
+
+func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) {
+	msgs, err := tmq.Ch.Consume(
+		queueName,
+		"",
+		false,
+		false,
+		false,
+		false,
+		nil,
+	)
+
+	if err != nil {
+		log.Printf("注册消费失败")
+		return
+	}
+
+	go func() {
+		for msg := range msgs {
+			var event rabbitmq.TimelineEvent
+			err := json.Unmarshal(msg.Body, &event)
+
+			if err != nil {
+				log.Printf("反序列化失败")
+				msg.Ack(false)
+				continue
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+			timelineKey := "feed:global_timeline"
+			err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{
+				Score:  float64(event.CreateTime),
+				Member: fmt.Sprintf("%d", event.VideoID),
+			})
+
+			if err != nil {
+				log.Printf("写入Zset失败")
+				msg.Nack(false, true)
+				cancel()
+				continue
+			}
+
+			err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001)
+
+			if err != nil {
+				log.Printf("ZRem失败")
+			}
+
+			msg.Ack(false)
+			cancel()
+		}
+	}()
+}