feat kafka 改版 2

db/kafka/consumer.go

@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"git.hlsq.asia/mmorpg/service-common/log"
@@ -27,13 +28,13 @@ type Consumer struct {
 	cancel context.CancelFunc
 }
 
-func (c *Consumer) Consume(h map[Topic]Handler) {
+func (c *Consumer) Consume(t []Topic) {
 	for {
 		topicArr := make([]string, 0)
-		handlerMap := make(map[string]Handler)
-		for t, h2 := range h {
-			topicArr = append(topicArr, t.Name())
-			handlerMap[t.Name()] = h2
+		handlerMap := make(map[string]Topic)
+		for _, v := range t {
+			topicArr = append(topicArr, v.Name())
+			handlerMap[v.Name()] = v
 		}
 		err := client.consumer.Consume(c.ctx, topicArr, &handler{
 			handler: handlerMap,
@@ -53,7 +54,7 @@ func (c *Consumer) Stop() error {
 type Handler func(context.Context, []byte) error
 
 type handler struct {
-	handler map[string]Handler
+	handler map[string]Topic
 }
 
 func (h *handler) Setup(_ sarama.ConsumerGroupSession) error {
@@ -77,7 +78,13 @@ func (h *handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.Co
 			return utils.ErrorsWrap(errors.New("handler not found"))
 		}
 
-		if err := cb(ctx, message.Value); err != nil {
+		if err := json.Unmarshal(message.Value, cb); err != nil {
+			span.SetStatus(otelcodes.Error, "handler json.Unmarshal error")
+			span.End()
+			return utils.ErrorsWrap(err, "handler json.Unmarshal error")
+		}
+
+		if err := cb.OnMessage(ctx); err != nil {
 			if stack, ok := err.(interface{ StackTrace() string }); ok {
 				span.AddEvent("Stack Trace", trace.WithAttributes(
 					attribute.String("stack.trace", fmt.Sprintf("%v", stack.StackTrace())),