feat: 流行度相关rabbitMQ

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)
+}
+

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)
+}

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
+}
+