290 lines
9.2 KiB
Go
290 lines
9.2 KiB
Go
|
package mq
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"context"
|
|||
|
|
"fmt"
|
|||
|
|
"github.com/apache/rocketmq-client-go/v2"
|
|||
|
|
"github.com/apache/rocketmq-client-go/v2/consumer"
|
|||
|
|
"github.com/apache/rocketmq-client-go/v2/primitive"
|
|||
|
|
"github.com/pkg/errors"
|
|||
|
|
"go.opentelemetry.io/otel"
|
|||
|
|
"go.opentelemetry.io/otel/codes"
|
|||
|
|
"go.opentelemetry.io/otel/propagation"
|
|||
|
|
semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
|
|||
|
|
"go.opentelemetry.io/otel/trace"
|
|||
|
|
"os"
|
|||
|
|
"strings"
|
|||
|
|
"sync/atomic"
|
|||
|
|
"time"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
// ConsumerManager 消费者管理器
|
|||
|
|
type ConsumerManager struct {
|
|||
|
|
consumerConfigs []*ConsumerConfig
|
|||
|
|
activeCnt atomic.Int32 //当前正在处理业务的个数，用于退出时业务平滑的退出
|
|||
|
|
shutdownFlag atomic.Bool // 关闭标记
|
|||
|
|
|
|||
|
|
logger Logger
|
|||
|
|
IsOpenTelemetry bool // 默认开启链路追踪，及时未配置上报地址也不影响业务，只是实际不会上报上去而已
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// NewConsumerManager 创建一个消费者管理器
|
|||
|
|
func NewConsumerManager(logger Logger) *ConsumerManager {
|
|||
|
|
return &ConsumerManager{logger: logger, IsOpenTelemetry: true}
|
|||
|
|
}
|
|||
|
|
func (c *ConsumerManager) DisableOpenTelemetry() {
|
|||
|
|
c.IsOpenTelemetry = false
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Subscribe 订阅一个主题
|
|||
|
|
func (c *ConsumerManager) Subscribe(ctx context.Context, connConf *ConsumerConnConfig, consumerConf *ConsumerConfig, fn func(context.Context, *ConsumerMessage) error, opts ...consumer.Option) error {
|
|||
|
|
// 检查参数规范
|
|||
|
|
if !c.checkGroupName(consumerConf.TopicName, consumerConf.GroupName) {
|
|||
|
|
return fmt.Errorf("groupName不符合规范，前缀必须是\"${topicName}_\"开头，如trade_recharge_dispatcher_pay")
|
|||
|
|
}
|
|||
|
|
if consumerConf.RetryCnt == nil || *consumerConf.RetryCnt < 0 {
|
|||
|
|
retryCnt := 38
|
|||
|
|
consumerConf.RetryCnt = &retryCnt
|
|||
|
|
}
|
|||
|
|
if consumerConf.ConsumerCnt <= 0 {
|
|||
|
|
consumerConf.ConsumerCnt = 1
|
|||
|
|
}
|
|||
|
|
if consumerConf.PerCoroutineCnt <= 0 {
|
|||
|
|
consumerConf.PerCoroutineCnt = 20
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
credentials := primitive.Credentials{
|
|||
|
|
AccessKey: connConf.AccessKey,
|
|||
|
|
SecretKey: connConf.SecretKey,
|
|||
|
|
SecurityToken: connConf.SecurityToken,
|
|||
|
|
}
|
|||
|
|
//限制groupName名称必须与topic捆绑
|
|||
|
|
nameServers := strings.Split(connConf.NameServers, ",")
|
|||
|
|
opts = append(opts,
|
|||
|
|
consumer.WithNameServer(nameServers),
|
|||
|
|
consumer.WithCredentials(credentials),
|
|||
|
|
consumer.WithConsumerModel(consumer.Clustering),
|
|||
|
|
// 不要开启此参数，开启顺序消费，通过返回consumer.ConsumeRetryLater将会失效，需要自己实现重试机制
|
|||
|
|
//consumer.WithConsumerOrder(true),
|
|||
|
|
consumer.WithGroupName(consumerConf.GroupName),
|
|||
|
|
consumer.WithConsumeGoroutineNums(consumerConf.PerCoroutineCnt),
|
|||
|
|
consumer.WithRetry(*consumerConf.RetryCnt),
|
|||
|
|
// 不启用批消费，多个一起消费的特点是则其中一个失败则整体都会失败
|
|||
|
|
consumer.WithConsumeMessageBatchMaxSize(1),
|
|||
|
|
//consumer.WithPullThresholdForQueue(50),
|
|||
|
|
)
|
|||
|
|
// 启动指定个数的consumer
|
|||
|
|
hostName, _ := os.Hostname()
|
|||
|
|
now := time.Now().Unix()
|
|||
|
|
for i := 0; i < consumerConf.ConsumerCnt; i++ {
|
|||
|
|
currOpts := make([]consumer.Option, len(opts))
|
|||
|
|
copy(currOpts, opts)
|
|||
|
|
currOpts = append(currOpts, consumer.WithInstance(fmt.Sprintf("%s:%s:%d:%d", hostName, consumerConf.GroupName, now, i+1)))
|
|||
|
|
pushConsumer, err := rocketmq.NewPushConsumer(currOpts...)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
selector := consumer.MessageSelector{}
|
|||
|
|
// 过滤tag
|
|||
|
|
if len(consumerConf.Tags) > 0 {
|
|||
|
|
selector.Type = consumer.TAG
|
|||
|
|
selector.Expression = strings.Join(consumerConf.Tags, " || ")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
err = pushConsumer.Subscribe(consumerConf.TopicName, selector, func(subCtx context.Context, ext ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
|
|||
|
|
return c.callbackForReceive(subCtx, consumerConf, fn, ext...)
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
consumerConf.pushConsumers = append(consumerConf.pushConsumers, pushConsumer)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
c.consumerConfigs = append(c.consumerConfigs, consumerConf)
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var consumerPropagator = propagation.TraceContext{}
|
|||
|
|
|
|||
|
|
// callbackForReceive 收到消息的回调
|
|||
|
|
func (c *ConsumerManager) callbackForReceive(ctx context.Context, consumerConf *ConsumerConfig, fn func(context.Context, *ConsumerMessage) error, ext ...*primitive.MessageExt) (cr consumer.ConsumeResult, fnErr error) {
|
|||
|
|
// 收到消息
|
|||
|
|
if c.shutdownFlag.Load() {
|
|||
|
|
cr = consumer.ConsumeRetryLater
|
|||
|
|
fnErr = fmt.Errorf("正在退出中，延期处理:%s,%s", consumerConf.TopicName, consumerConf.GroupName)
|
|||
|
|
// 卡住，不再继续消费，等待退出
|
|||
|
|
// 测试发现在重试主题消费时，返回retryLater有时会被commit掉，导致消息丢失
|
|||
|
|
time.Sleep(24 * time.Hour)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 标记活跃状态
|
|||
|
|
c.activeCnt.Add(1)
|
|||
|
|
defer func() {
|
|||
|
|
c.activeCnt.Add(-1)
|
|||
|
|
if v := recover(); v != nil {
|
|||
|
|
cr = consumer.ConsumeRetryLater
|
|||
|
|
fnErr = errors.Errorf("处理消息panic, groupName=%s，%+v", consumerConf.GroupName, v)
|
|||
|
|
c.LogErrorf("%+v", fnErr)
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
|
|||
|
|
var tracer trace.Tracer
|
|||
|
|
if c.IsOpenTelemetry {
|
|||
|
|
tracer = otel.GetTracerProvider().Tracer("LSXD_Util")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// WithConsumeMessageBatchMaxSize 配置大于1时，ext才会存在多个，它的特点是要么全成功或全失败
|
|||
|
|
cr = consumer.ConsumeSuccess
|
|||
|
|
spanName := fmt.Sprintf("%s %s %s", consumerConf.TopicName, semconv.MessagingOperationProcess.Value.AsString(), consumerConf.GroupName)
|
|||
|
|
for _, v := range ext {
|
|||
|
|
func() {
|
|||
|
|
message := &ConsumerMessage{
|
|||
|
|
MsgId: v.MsgId,
|
|||
|
|
Topic: v.Topic,
|
|||
|
|
Body: v.Body,
|
|||
|
|
ReconsumeTimes: v.ReconsumeTimes,
|
|||
|
|
CompressedBody: v.CompressedBody,
|
|||
|
|
Flag: v.Flag,
|
|||
|
|
TransactionId: v.TransactionId,
|
|||
|
|
Batch: v.Batch,
|
|||
|
|
Compress: v.Compress,
|
|||
|
|
Properties: v.GetProperties(),
|
|||
|
|
}
|
|||
|
|
// 链路追踪
|
|||
|
|
var span trace.Span
|
|||
|
|
if tracer != nil {
|
|||
|
|
if traceParent, ok := message.Properties[openTelemetryPropertyName]; ok {
|
|||
|
|
var mapCarrier propagation.MapCarrier = map[string]string{
|
|||
|
|
openTelemetryPropertyName: traceParent,
|
|||
|
|
}
|
|||
|
|
ctx = consumerPropagator.Extract(ctx, mapCarrier)
|
|||
|
|
}
|
|||
|
|
ctx, span = tracer.Start(ctx, spanName, trace.WithSpanKind(trace.SpanKindConsumer))
|
|||
|
|
span.SetAttributes(
|
|||
|
|
semconv.MessagingSystem("RocketMQ"),
|
|||
|
|
semconv.MessagingSourceName(consumerConf.TopicName),
|
|||
|
|
semconv.MessagingRocketmqClientGroup(consumerConf.GroupName),
|
|||
|
|
semconv.MessagingRocketmqMessageKeys(message.GetKeys()...),
|
|||
|
|
semconv.MessagingRocketmqMessageTag(message.GetTags()),
|
|||
|
|
semconv.MessagingRocketmqMessageDelayTimeLevel(message.GetDelayTimeLevel()),
|
|||
|
|
semconv.MessageIDKey.String(message.MsgId),
|
|||
|
|
)
|
|||
|
|
defer func() {
|
|||
|
|
// 记录追踪信息
|
|||
|
|
spanErr := fnErr
|
|||
|
|
var panicVal any
|
|||
|
|
if panicVal = recover(); panicVal != nil {
|
|||
|
|
spanErr = fmt.Errorf("%s消费者处理方法panic:%s", consumerConf.GroupName, panicVal)
|
|||
|
|
}
|
|||
|
|
if spanErr != nil {
|
|||
|
|
span.RecordError(spanErr)
|
|||
|
|
span.SetStatus(codes.Error, spanErr.Error())
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
span.End()
|
|||
|
|
|
|||
|
|
if panicVal != nil {
|
|||
|
|
panic(panicVal)
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// 回调业务函数
|
|||
|
|
currErr := fn(ctx, message)
|
|||
|
|
|
|||
|
|
if currErr != nil {
|
|||
|
|
cr = consumer.ConsumeRetryLater
|
|||
|
|
fnErr = currErr
|
|||
|
|
c.LogErrorf("%s消费者处理方法返回失败,body=%s：%+v", consumerConf.GroupName, string(message.Body), currErr)
|
|||
|
|
}
|
|||
|
|
}()
|
|||
|
|
}
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Start 启动所有消费者
|
|||
|
|
func (c *ConsumerManager) Start(_ context.Context) error {
|
|||
|
|
for _, consumerConf := range c.consumerConfigs {
|
|||
|
|
for _, pushConsumer := range consumerConf.pushConsumers {
|
|||
|
|
err := pushConsumer.Start()
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Stop 停止所有消费者
|
|||
|
|
func (c *ConsumerManager) Stop(_ context.Context) error {
|
|||
|
|
fmt.Println("开始停止消费者")
|
|||
|
|
c.shutdownFlag.Store(true)
|
|||
|
|
|
|||
|
|
for _, consumerConf := range c.consumerConfigs {
|
|||
|
|
for _, pushConsumer := range consumerConf.pushConsumers {
|
|||
|
|
pushConsumer.Suspend() // 似乎没起使用
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
fmt.Println("已suspend所有消费者")
|
|||
|
|
|
|||
|
|
//shutdown之间，保证正在处理的消费先提交
|
|||
|
|
_ = c.blockWaitFinish()
|
|||
|
|
|
|||
|
|
var err error = nil
|
|||
|
|
for _, consumerConf := range c.consumerConfigs {
|
|||
|
|
for _, pushConsumer := range consumerConf.pushConsumers {
|
|||
|
|
if closeErr := pushConsumer.Shutdown(); closeErr != nil {
|
|||
|
|
err = closeErr
|
|||
|
|
fmt.Println("消费者shutdown失败", closeErr)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
fmt.Println("已shutdown所有消费者")
|
|||
|
|
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// blockWaitFinish 阻塞等待业务完成
|
|||
|
|
func (c *ConsumerManager) blockWaitFinish() error {
|
|||
|
|
// 每1s检查下业务是否都处理完成
|
|||
|
|
|
|||
|
|
for {
|
|||
|
|
cnt := c.activeCnt.Load()
|
|||
|
|
if cnt == 0 {
|
|||
|
|
//无业务处理，正常退
|
|||
|
|
break
|
|||
|
|
} else {
|
|||
|
|
fmt.Printf("等待消费者退出，%d 个正在运行\n", cnt)
|
|||
|
|
}
|
|||
|
|
time.Sleep(1 * time.Second)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
//防止极端情况下commit未完成
|
|||
|
|
// nolint
|
|||
|
|
time.Sleep(1 * time.Second)
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// LogErrorf 记录错误日志
|
|||
|
|
func (c *ConsumerManager) LogErrorf(format string, args ...any) {
|
|||
|
|
if c.logger != nil {
|
|||
|
|
c.logger.Errorf(format, args...)
|
|||
|
|
} else {
|
|||
|
|
fmt.Printf(format+"\n", args...)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// checkGroupName 检查groupName是否符合规范
|
|||
|
|
// 必须是以${topicName}_开头，目的是
|
|||
|
|
// 1. 防止相同相同groupName与多个topic的情况，避免出现消费不符异常的情况
|
|||
|
|
// 2. 在管理group时，能更清晰地体现出对应的topic
|
|||
|
|
func (c *ConsumerManager) checkGroupName(topicName string, groupName string) bool {
|
|||
|
|
if groupName == "" {
|
|||
|
|
return false
|
|||
|
|
}
|
|||
|
|
return strings.HasPrefix(groupName, topicName+"_")
|
|||
|
|
}
|