114 lines
3.6 KiB
Go
114 lines
3.6 KiB
Go
package yumigiftchallenge
|
||
|
||
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 异步启动 Yumi 独立消费组；MQ 临时不可用不能阻断 API 启动。
|
||
func (s *Service) StartMessageConsumer(ctx context.Context) error {
|
||
if !s.cfg.YumiGiftChallenge.RocketMQ.Enabled {
|
||
log.Printf("yumi gift challenge rocketmq consumer disabled")
|
||
return nil
|
||
}
|
||
mqConfig := s.cfg.YumiGiftChallenge.RocketMQ
|
||
if strings.TrimSpace(mqConfig.Endpoint) == "" || strings.TrimSpace(mqConfig.AccessKey) == "" || strings.TrimSpace(mqConfig.AccessSecret) == "" {
|
||
log.Printf("yumi gift challenge rocketmq consumer skipped: endpoint or credentials missing")
|
||
return nil
|
||
}
|
||
go s.startRocketMQConsumerWithRetry(ctx)
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) startRocketMQConsumerWithRetry(ctx context.Context) {
|
||
retryDelay := 5 * time.Second
|
||
for ctx.Err() == nil {
|
||
if err := s.startRocketMQConsumer(ctx); err == nil {
|
||
return
|
||
} else {
|
||
log.Printf("start yumi gift challenge 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
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) startRocketMQConsumer(ctx context.Context) error {
|
||
mqConfig := s.cfg.YumiGiftChallenge.RocketMQ
|
||
filter := rmq.SUB_ALL
|
||
if tag := strings.TrimSpace(mqConfig.Tag); tag != "" {
|
||
filter = rmq.NewFilterExpression(tag)
|
||
}
|
||
consumer, err := rmq.NewPushConsumer(&rmq.Config{
|
||
Endpoint: mqConfig.Endpoint, NameSpace: mqConfig.Namespace, ConsumerGroup: mqConfig.ConsumerGroup,
|
||
Credentials: &credentials.SessionCredentials{
|
||
AccessKey: mqConfig.AccessKey, AccessSecret: mqConfig.AccessSecret, SecurityToken: mqConfig.SecurityToken,
|
||
},
|
||
},
|
||
rmq.WithPushSubscriptionExpressions(map[string]*rmq.FilterExpression{mqConfig.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
|
||
}
|
||
if err := s.processRocketMQMessage(context.Background(), messageView); err != nil {
|
||
log.Printf("yumi gift challenge 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("yumi gift challenge rocketmq consumer started. topic=%s group=%s tag=%s", mqConfig.Topic, mqConfig.ConsumerGroup, mqConfig.Tag)
|
||
go func() {
|
||
<-ctx.Done()
|
||
if s.rocketConsumer != nil {
|
||
_ = s.rocketConsumer.GracefulStop()
|
||
}
|
||
}()
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) 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.Is(err, errMalformedGiftPayload) || errors.As(err, &syntaxErr) || errors.As(err, &typeErr) {
|
||
// 毒消息重试不会自愈；记录 messageId 后 ACK，数据库或下游错误仍返回 FAILURE 重试。
|
||
log.Printf("drop malformed yumi gift challenge message. messageId=%s err=%v", messageView.GetMessageId(), err)
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
return nil
|
||
}
|