diff --git a/backend/internal/middleware/rabbitmq/popularityMQ.go b/backend/internal/middleware/rabbitmq/popularityMQ.go new file mode 100644 index 0000000..4f37d4d --- /dev/null +++ b/backend/internal/middleware/rabbitmq/popularityMQ.go @@ -0,0 +1,57 @@ +package rabbitmq + +import ( + "context" + "errors" + "time" +) + +type PopularityMQ struct { + *RabbitMQ +} + +const ( + popularityExchange = "video.popularity.events" + popularityQueue = "video.popularity.events" + popularityBindingKey = "video.popularity.*" + + popularityUpdateRK = "video.popularity.update" +) + +type PopularityEvent struct { + EventID string `json:"event_id"` + VideoID uint `json:"video_id"` + Change int64 `json:"change"` + OccurredAt time.Time `json:"occurred_at"` +} + +func NewPopularityMQ(base *RabbitMQ) (*PopularityMQ, error) { + if base == nil { + return nil, errors.New("rabbitmq base is nil") + } + if err := base.DeclareTopic(popularityExchange, popularityQueue, popularityBindingKey); err != nil { + return nil, err + } + return &PopularityMQ{RabbitMQ: base}, nil +} + +func (p *PopularityMQ) Update(ctx context.Context, videoID uint, change int64) error { + if p == nil || p.RabbitMQ == nil { + return errors.New("popularity mq is not initialized") + } + if videoID == 0 || change == 0 { + return errors.New("videoID and change are required") + } + id, err := newEventID(16) + if err != nil { + return err + } + event := PopularityEvent{ + EventID: id, + VideoID: videoID, + Change: change, + OccurredAt: time.Now().UTC(), + } + return p.PublishJSON(ctx, popularityExchange, popularityUpdateRK, event) +} + diff --git a/backend/internal/video/popularity_cache.go b/backend/internal/video/popularity_cache.go new file mode 100644 index 0000000..0117bc0 --- /dev/null +++ b/backend/internal/video/popularity_cache.go @@ -0,0 +1,29 @@ +package video + +import ( + "context" + "fmt" + "strconv" + "time" + + rediscache "feedsystem_video_go/internal/middleware/redis" +) + +// 更新视频流行度缓存 +func UpdatePopularityCache(ctx context.Context, cache *rediscache.Client, id uint, change int64) { + if cache == nil || id == 0 || change == 0 { + return + } + + _ = cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id)) + + now := time.Now().UTC().Truncate(time.Minute) + windowKey := "hot:video:1m:" + now.Format("200601021504") + member := strconv.FormatUint(uint64(id), 10) + + opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + _ = cache.ZincrBy(opCtx, windowKey, member, float64(change)) + _ = cache.Expire(opCtx, windowKey, 2*time.Hour) +} diff --git a/backend/internal/worker/popularityworker.go b/backend/internal/worker/popularityworker.go new file mode 100644 index 0000000..8184425 --- /dev/null +++ b/backend/internal/worker/popularityworker.go @@ -0,0 +1,79 @@ +package worker + +import ( + "context" + "encoding/json" + "errors" + "feedsystem_video_go/internal/middleware/rabbitmq" + rediscache "feedsystem_video_go/internal/middleware/redis" + "feedsystem_video_go/internal/video" + "log" + + amqp "github.com/rabbitmq/amqp091-go" +) + +type PopularityWorker struct { + ch *amqp.Channel + cache *rediscache.Client + queue string +} + +func NewPopularityWorker(ch *amqp.Channel, cache *rediscache.Client, queue string) *PopularityWorker { + return &PopularityWorker{ch: ch, cache: cache, queue: queue} +} + +func (w *PopularityWorker) Run(ctx context.Context) error { + if w == nil || w.ch == nil || w.cache == nil { + return errors.New("popularity worker is not initialized") + } + if w.queue == "" { + return errors.New("queue is required") + } + + deliveries, err := w.ch.Consume( + w.queue, + "", + false, + false, + false, + false, + nil, + ) + if err != nil { + return err + } + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case d, ok := <-deliveries: + if !ok { + return errors.New("deliveries channel closed") + } + w.handleDelivery(ctx, d) + } + } +} + +func (w *PopularityWorker) handleDelivery(ctx context.Context, d amqp.Delivery) { + if err := w.process(ctx, d.Body); err != nil { + log.Printf("popularity worker: failed to process message: %v", err) + _ = d.Nack(false, true) + return + } + _ = d.Ack(false) +} + +func (w *PopularityWorker) process(ctx context.Context, body []byte) error { + var evt rabbitmq.PopularityEvent + if err := json.Unmarshal(body, &evt); err != nil { + return nil + } + if evt.VideoID == 0 || evt.Change == 0 { + return nil + } + video.UpdatePopularityCache(ctx, w.cache, evt.VideoID, evt.Change) + return nil +} +