• 腾讯mini项目-【指标监控服务重构】2023-08-16

    今日已办

    v1

    验证 StageHandler 在处理消息时是否为单例，【错误尝试】

    type StageHandler struct {
}

func (s StageHandler) Middleware1(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		log.Logger.Info("StageHandler Middleware 1")
		fmt.Printf("%p\n", &s)
		return h(msg)
	}
}

func (s StageHandler) Middleware2(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		log.Logger.Info("StageHandler Middleware 2")
		fmt.Printf("%p\n", &s)
		return h(msg)
	}
}

func (s StageHandler) Middleware3(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		log.Logger.Info("StageHandler Middleware 3")
		fmt.Printf("%p\n", &s)
		return h(msg)
	}
}

func (s StageHandler) Handler1(msg *message.Message) error {
	log.Logger.Info("StageHandler Handler 1")
	fmt.Printf("%p\n", &s)
	return nil
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

    image-20230816161140984

    v2

    • 定义不同 Handler
    type CrashHandler struct {
	Topic string
}

func (s CrashHandler) Handler1(msg *message.Message) error {
	log.Logger.Info(s.Topic + ": CrashHandler Handler 1 start")
	fmt.Printf("%p\n", &s)
	time.Sleep(1 * time.Second)
	log.Logger.Info(s.Topic + ": CrashHandler Handler 1 end")
	return nil
}

type LagHandler struct {
	Topic string
}

func (s LagHandler) Handler1(msg *message.Message) error {
	log.Logger.Info(s.Topic + ": LagHandler Handler 1 start")
	fmt.Printf("%p\n", &s)
	time.Sleep(1 * time.Second)
	log.Logger.Info(s.Topic + ": LagHandler Handler 1 end")
	return nil
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 添加到router中
    	for _, topic := range topics {
		var category string
		var handlerFunc message.NoPublishHandlerFunc
		if strings.Contains(topic, performance.CategoryCrash) {
			category = performance.CategoryCrash
			handlerFunc = CrashHandler{Topic: category}.Handler1
		} else if strings.Contains(topic, performance.CategoryLag) {
			category = performance.CategoryLag
			handlerFunc = LagHandler{Topic: category}.Handler1
		} else {
			continue
		}

		handler := router.AddNoPublisherHandler(topic+"test-handler", topic, subscriber, handlerFunc)
	}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    image-20230816171659632

    • 结论
      • handler 实例会不断创建
      • 不同的 handler 可以并行处理不同主题的消息
      • 相同的 handler 在处理该主题的消息时是顺序的

    官方文档： Message Router (watermill.io)

    订阅者可以一次消费一条消息，也可以并行消费多条消息

    • Single stream of messages 是最简单的方法，这意味着在调用 msg.Ack() 之前，订阅者将不会收到任何新消息
    • Multiple message streams 仅部分订阅者支持。通过一次订阅多个主题分区，可以并行消费多条消息，甚至是之前未确认的消息（例如，Kafka 订阅者就是这样工作的） Router 通过运行并发 HandlerFuncs（每个分区一个）来处理此模型

    v3

    存在并发安全问题

    1. 公用一个上下文
    2. 频繁的修改上下文中的字段值
    3. 不同Handler和MiddleWare存在并发

    解决思路

    • 将一次消息处理会使用到的数据集合定义为一个结构体
    type ContextData struct {
	Status int
	Event  schema.Event

	AppID         string // API 上报
	FetchScenario string // API 上报
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 使用message的Context来传递这个数据

    image-20230816222038683

    • 移除掉 ProfileCtx 的相关设计
    • 使用watermillzap.Logger来替换本身的 LoggerAdapter，更加直观且与原项目适配image-20230816225840032

    完整代码

    profile/internal/watermill/consumer/consumer_context.go

    // Package consumer
// @Author xzx 2023/8/11 18:53:00
package consumer

import (
	"context"
	kc "github.com/Kevinello/kafka-client"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/ThreeDotsLabs/watermill/message/router/plugin"
	"github.com/garsue/watermillzap"
	"github.com/qiulin/watermill-kafkago/pkg/kafkago"
	"go.uber.org/zap"
	"profile/internal/config"
	"profile/internal/connector"
	"profile/internal/log"
	"profile/internal/schema/performance"
	"strings"
	"time"
)

// Consume
// @Description
// @Author xzx 2023-08-16 22:52:52
func Consume() {
	logger := watermillzap.NewLogger(log.Logger)

	publisher, subscriber := newPubSub(logger)
	router, err := message.NewRouter(message.RouterConfig{}, logger)
	if err != nil {
		log.Logger.Fatal("creates a new Router with given configuration error", zap.Error(err))
	}

	router.AddPlugin(plugin.SignalsHandler)
	router.AddMiddleware(
		middleware.Retry{
			MaxRetries:      3,
			InitialInterval: time.Millisecond * 100,
			Logger:          logger,
		}.Middleware,
		middleware.Recoverer,
	)

	getTopics := kc.GetTopicReMatch(strings.Split(config.Profile.GetString("kafka.topicRE"), ","))
	topics, err := getTopics(config.Profile.GetString("kafka.bootstrap"))
	if err != nil {
		log.Logger.Fatal("get topics failed", zap.Error(err))
		return
	}

	for _, topic := range topics {
		var category string
		var handlerFunc message.HandlerFunc
		if strings.Contains(topic, performance.CategoryCrash) {
			category = performance.CategoryCrash
			handlerFunc = CrashWriteKafka
		} else if strings.Contains(topic, performance.CategoryLag) {
			category = performance.CategoryLag
			handlerFunc = LagWriteKafka
		} else {
			continue
		}
		router.AddHandler(category, topic, subscriber, connector.GetTopic(category), publisher, handlerFunc).
			AddMiddleware(
				UnpackKafkaMessage,
				InitPerformanceEvent,
				AnalyzeEvent)
	}

	if err = router.Run(context.Background()); err != nil {
		log.Logger.Error("runs all plugins and handlers and starts subscribing to provided topics error", zap.Error(err))
	}
}

// newPubSub
// @Description
// @Author xzx 2023-08-16 22:52:45
// @Param logger
// @Return message.Publisher
// @Return message.Subscriber
func newPubSub(logger watermill.LoggerAdapter) (message.Publisher, message.Subscriber) {
	marshaler := kafkago.DefaultMarshaler{}
	publisher := kafkago.NewPublisher(kafkago.PublisherConfig{
		Brokers:     []string{config.Profile.GetString("kafka.bootstrap")},
		Async:       false,
		Marshaler:   marshaler,
		OTELEnabled: false,
		Ipv4Only:    true,
		Timeout:     100 * time.Second,
	}, logger)

	subscriber, err := kafkago.NewSubscriber(kafkago.SubscriberConfig{
		Brokers:       []string{config.Profile.GetString("kafka.bootstrap")},
		Unmarshaler:   marshaler,
		ConsumerGroup: config.Profile.GetString("kafka.group"),
		OTELEnabled:   false,
	}, logger)
	if err != nil {
		log.Logger.Fatal("Unable to create subscriber", zap.Error(err))
	}
	return publisher, subscriber
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103

    profile/internal/watermill/consumer/consumer_stage.go

    // Package consumer
// @Author xzx 2023/8/12 10:01:00
package consumer

import (
	"context"
	"encoding/json"
	"github.com/ThreeDotsLabs/watermill/message"
	"go.uber.org/zap"
	"profile/internal/connector"
	"profile/internal/log"
	"profile/internal/schema"
	"profile/internal/schema/performance"
	"profile/internal/state"
)

type ContextData struct {
	Status int
	Event  schema.Event

	AppID         string // API 上报
	FetchScenario string // API 上报
}

// UnpackKafkaMessage
// @Description
// @Author xzx 2023-08-12 12:27:30
// @Param h
// @Return message.HandlerFunc
func UnpackKafkaMessage(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		var data ContextData
		// 反序列化，存入通用结构体
		if contextErr := json.Unmarshal(msg.Payload, &data.Event); contextErr != nil {
			data.Status = state.StatusUnmarshalError
			return nil, contextErr
		}
		log.Logger.Info("[1-UnpackKafkaItem] unpack kafka item success", zap.Any("event", data.Event))

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		return h(msg)
	}
}

// InitPerformanceEvent
// @Description
// @Author xzx 2023-08-12 12:27:35
// @Param h
// @Return message.HandlerFunc
func InitPerformanceEvent(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		data := msg.Context().Value("data").(ContextData)
		event, contextErr := performance.EventFactory(data.Event.Category, data.Event.Dimensions, data.Event.Values)
		if contextErr != nil {
			data.Status = state.StatusEventFactoryError
			return nil, contextErr
		}
		log.Logger.Info("[2-InitPerformanceEvent] Consume performance event success", zap.Any("event", data.Event))
		data.Event.ProfileData = event

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		return h(msg)
	}
}

// AnalyzeEvent
// @Description
// @Author xzx 2023-08-12 12:27:38
// @Param h
// @Return message.HandlerFunc
func AnalyzeEvent(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		data := msg.Context().Value("data").(ContextData)

		contextErr := data.Event.ProfileData.Analyze()
		if contextErr != nil {
			data.Status = state.StatusAnalyzeError
			return nil, contextErr
		}
		log.Logger.Info("[3-AnalyzeEvent] analyze event success", zap.Any("event", data.Event))
		// clear dimensions and values
		data.Event.Dimensions = nil
		data.Event.Values = nil

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		return h(msg)
	}
}

// CrashWriteKafka
// @Description
// @Author xzx 2023-08-12 15:09:15
// @Param msg
// @Return []*message.Message
// @Return error
func CrashWriteKafka(msg *message.Message) ([]*message.Message, error) {
	data := msg.Context().Value("data").(ContextData)

	toWriteBytes, contextErr := json.Marshal(data.Event)
	if contextErr != nil {
		data.Status = state.StatusUnmarshalError
		return nil, contextErr
	}

	msg = message.NewMessage(data.Event.ID, toWriteBytes)

	log.Logger.Info("[4-CrashWriteKafka] write kafka success", zap.String("topic", connector.GetTopic(data.Event.Category)), zap.String("id", data.Event.ID), zap.String("msg", string(toWriteBytes)))
	return message.Messages{msg}, nil
}

func LagWriteKafka(msg *message.Message) ([]*message.Message, error) {
	data := msg.Context().Value("data").(ContextData)

	toWriteBytes, contextErr := json.Marshal(data.Event)
	if contextErr != nil {
		data.Status = state.StatusUnmarshalError
		return nil, contextErr
	}

	msg = message.NewMessage(data.Event.ID, toWriteBytes)

	log.Logger.Info("[4-LagWriteKafka] write kafka success", zap.String("topic", connector.GetTopic(data.Event.Category)), zap.String("id", data.Event.ID), zap.String("msg", string(toWriteBytes)))
	return message.Messages{msg}, nil
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125

    测试

    上报PERF_LAG Event可以并发处理 2 条消息，不必等待上一条消息处理完

    image-20230816222712201

    image-20230816222820505

    多次测试发现是由于两条消息走了不同的 Handler

    image-20230816230158777

    暂未修复，明明是同一主题的两条消息却都走了两条不同的链路，而且 publisher 最后写回的主题也是写到了不同的主题上，并且上报另一个类型的事件，即另一个主题的消息却无法触发消费者消费！

    暂定先写死两个主题名称测试是否正常

    明日待办

    1. 开会讨论项目规划和任务分工
    2. 继续完成需求
  • 相关阅读:
    LiveData简单使用
    DTSE Tech Talk丨第3期：解密数据隔离方案，让SaaS应用开发更轻松
    包装流的使用练习
    项目成本管理
    mybatis-puls使用修改语句部分字段不生效
    手撕code
    如何保留 Excel 表头和第一行数据并追加 CSV 数据
    计算机丢失msvcr110.dll解决办法，多种msvcr110.dll解决方法分享
    apache paimon在flink中做维表join的优势
    STM32标准库（固件库）分析
  • 原文地址：https://blog.csdn.net/xzx18822942899/article/details/132913872