135 lines
3.9 KiB
Go
135 lines
3.9 KiB
Go
package weekstar
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
rmq "github.com/apache/rocketmq-clients/golang/v5"
|
||
"github.com/apache/rocketmq-clients/golang/v5/credentials"
|
||
)
|
||
|
||
// StartMessageConsumer 启动送礼消息消费链路。
|
||
func (s *WeekStarService) StartMessageConsumer(ctx context.Context) error {
|
||
if !s.cfg.WeekStar.RocketMQ.Enabled {
|
||
log.Printf("week star rocketmq consumer disabled")
|
||
return nil
|
||
}
|
||
if strings.TrimSpace(s.cfg.WeekStar.RocketMQ.Endpoint) == "" ||
|
||
strings.TrimSpace(s.cfg.WeekStar.RocketMQ.AccessKey) == "" ||
|
||
strings.TrimSpace(s.cfg.WeekStar.RocketMQ.AccessSecret) == "" {
|
||
log.Printf("week star rocketmq consumer skipped: endpoint or credentials missing")
|
||
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)
|
||
}
|
||
|
||
consumer, err := rmq.NewPushConsumer(&rmq.Config{
|
||
Endpoint: s.cfg.WeekStar.RocketMQ.Endpoint,
|
||
NameSpace: s.cfg.WeekStar.RocketMQ.Namespace,
|
||
ConsumerGroup: s.cfg.WeekStar.RocketMQ.ConsumerGroup,
|
||
Credentials: &credentials.SessionCredentials{
|
||
AccessKey: s.cfg.WeekStar.RocketMQ.AccessKey,
|
||
AccessSecret: s.cfg.WeekStar.RocketMQ.AccessSecret,
|
||
SecurityToken: s.cfg.WeekStar.RocketMQ.SecurityToken,
|
||
},
|
||
},
|
||
rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{
|
||
s.cfg.WeekStar.RocketMQ.Topic: filter,
|
||
}),
|
||
rmq.WithPushConsumptionThreadCount(8),
|
||
rmq.WithPushMaxCacheMessageCount(256),
|
||
rmq.WithPushMessageListener(&rmq.FuncMessageListener{
|
||
Consume: func(messageView *rmq.MessageView) rmq.ConsumerResult {
|
||
if messageView == nil {
|
||
return rmq.SUCCESS
|
||
}
|
||
// 单条消息处理失败时返回 FAILURE，让 RocketMQ 触发重试。
|
||
if err := s.processRocketMQMessage(context.Background(), messageView); err != nil {
|
||
log.Printf("week star mq process failed. messageId=%s err=%v", messageView.GetMessageId(), err)
|
||
return rmq.FAILURE
|
||
}
|
||
return rmq.SUCCESS
|
||
},
|
||
}),
|
||
)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := consumer.Start(); err != nil {
|
||
return err
|
||
}
|
||
s.rocketConsumer = consumer
|
||
log.Printf("week star rocketmq consumer started. topic=%s group=%s tag=%s",
|
||
s.cfg.WeekStar.RocketMQ.Topic,
|
||
s.cfg.WeekStar.RocketMQ.ConsumerGroup,
|
||
s.cfg.WeekStar.RocketMQ.Tag,
|
||
)
|
||
go func() {
|
||
<-ctx.Done()
|
||
if s.rocketConsumer != nil {
|
||
_ = s.rocketConsumer.GracefulStop()
|
||
}
|
||
}()
|
||
return nil
|
||
}
|
||
|
||
// processRocketMQMessage 解析 RocketMQ 消息体并交给业务逻辑处理。
|
||
func (s *WeekStarService) processRocketMQMessage(ctx context.Context, messageView *rmq.MessageView) error {
|
||
payload := strings.TrimSpace(string(messageView.GetBody()))
|
||
if payload == "" {
|
||
return nil
|
||
}
|
||
if err := s.ProcessGiftPayload(ctx, payload); err != nil {
|
||
var syntaxErr *json.SyntaxError
|
||
var typeErr *json.UnmarshalTypeError
|
||
if errors.As(err, &syntaxErr) || errors.As(err, &typeErr) {
|
||
log.Printf("drop malformed week star message. messageId=%s err=%v", messageView.GetMessageId(), err)
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
return nil
|
||
}
|