diff --git a/internal/service/weekstar/lifecycle.go b/internal/service/weekstar/lifecycle.go index bce6d22..30dd3da 100644 --- a/internal/service/weekstar/lifecycle.go +++ b/internal/service/weekstar/lifecycle.go @@ -6,6 +6,7 @@ import ( "errors" "log" "strings" + "time" rmq "github.com/apache/rocketmq-clients/golang/v5" "github.com/apache/rocketmq-clients/golang/v5/credentials" @@ -13,11 +14,6 @@ import ( // StartMessageConsumer 启动送礼消息消费链路。 func (s *WeekStarService) StartMessageConsumer(ctx context.Context) error { - return s.startRocketMQConsumer(ctx) -} - -// startRocketMQConsumer 启动 RocketMQ 推送消费者,订阅送礼事件。 -func (s *WeekStarService) startRocketMQConsumer(ctx context.Context) error { if !s.cfg.WeekStar.RocketMQ.Enabled { log.Printf("week star rocketmq consumer disabled") return nil @@ -29,6 +25,41 @@ func (s *WeekStarService) startRocketMQConsumer(ctx context.Context) error { return nil } + go s.startRocketMQConsumerWithRetry(ctx) + return nil +} + +// startRocketMQConsumerWithRetry 在后台持续尝试启动消费者,避免 MQ 配置或 topic 临时异常拖垮 HTTP API。 +func (s *WeekStarService) startRocketMQConsumerWithRetry(ctx context.Context) { + retryDelay := 5 * time.Second + for { + if ctx.Err() != nil { + return + } + + if err := s.startRocketMQConsumer(ctx); err != nil { + log.Printf("start week star rocketmq consumer failed: %v; retry in %s", err, retryDelay) + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + if retryDelay < time.Minute { + retryDelay *= 2 + if retryDelay > time.Minute { + retryDelay = time.Minute + } + } + continue + } + return + } +} + +// startRocketMQConsumer 启动 RocketMQ 推送消费者,订阅送礼事件。 +func (s *WeekStarService) startRocketMQConsumer(ctx context.Context) error { filter := rmq.SUB_ALL if tag := strings.TrimSpace(s.cfg.WeekStar.RocketMQ.Tag); tag != "" { filter = rmq.NewFilterExpression(tag) diff --git a/internal/service/weekstar/lifecycle_test.go b/internal/service/weekstar/lifecycle_test.go new file mode 100644 index 0000000..b730436 --- /dev/null +++ b/internal/service/weekstar/lifecycle_test.go @@ -0,0 +1,36 @@ +package weekstar + +import ( + "context" + "testing" + "time" + + "chatapp3-golang/internal/config" +) + +func TestStartMessageConsumerDoesNotBlockAPIStartup(t *testing.T) { + service := NewWeekStarService(config.Config{ + WeekStar: config.WeekStarConfig{ + RocketMQ: config.WeekStarRocketMQConfig{ + Enabled: true, + Endpoint: "127.0.0.1:1", + AccessKey: "test-access-key", + AccessSecret: "test-access-secret", + ConsumerGroup: "week-star-activity-test", + Topic: "GIVE_GIFTS", + Tag: "give_gift_v3", + }, + }, + }, nil, nil, nil) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + start := time.Now() + if err := service.StartMessageConsumer(ctx); err != nil { + t.Fatalf("StartMessageConsumer() error = %v", err) + } + if elapsed := time.Since(start); elapsed > 100*time.Millisecond { + t.Fatalf("StartMessageConsumer() blocked for %s", elapsed) + } +}