From d76a11f62fbd2b49686abb6f06ba5b2a598e8450 Mon Sep 17 00:00:00 2001
From: hy001 <hy@hy001deMacBook-Pro.local>
Date: Thu, 30 Apr 2026 20:33:44 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Drocket=E6=8A=A5=E9=94=99?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/service/weekstar/lifecycle.go      | 41 ++++++++++++++++++---
 internal/service/weekstar/lifecycle_test.go | 36 ++++++++++++++++++
 2 files changed, 72 insertions(+), 5 deletions(-)
 create mode 100644 internal/service/weekstar/lifecycle_test.go

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