feat kafka 改版 1

internal/timer/consumer.go

@@ -9,7 +9,6 @@ import (
 	"git.hlsq.asia/mmorpg/service-common/utils"
 	"git.hlsq.asia/mmorpg/service-qgdzs/internal/dao/model"
 	"git.hlsq.asia/mmorpg/service-qgdzs/internal/dao/repository"
-	"github.com/IBM/sarama"
 	"gorm.io/gorm"
 )
 
@@ -21,31 +20,89 @@ type TopicQuestionAnswer struct {
 	Answer         string `json:"answer"`
 }
 
-func startConsumer() {
-	kafka.NewConsumer().Consume("qgdzs.question.answer", func(ctx context.Context, msg *sarama.ConsumerMessage) error {
-		log.Infof("Kafka topic: qgdzs.question.answer: %s", string(msg.Value))
-		data := &TopicQuestionAnswer{}
-		if err := json.Unmarshal(msg.Value, &data); err != nil {
-			return utils.ErrorsWrap(err)
+func (t *TopicQuestionAnswer) Name() string {
+	return "qgdzs.question.answer"
 }
+
+type TopicAddPoint struct {
+	MessageSn string         `json:"messageSn"`
+	Source    AddPointSource `json:"source"`
+	USN       string         `json:"usn"`
+	Point     int64          `json:"point"`
+}
+
+type AddPointSource int32
+
+const (
+	AddPointSourceUnKnown  = 0
+	AddPointSourceRandom   = 1
+	AddPointSourceCategory = 2
+	AddPointSourceQuickly  = 3
+)
+
+func (t *TopicAddPoint) Name() string {
+	return "qgdzs.user.point"
+}
+
+// ！！！消费者必须做幂等！！！
+func startConsumer() {
+	go kafka.NewConsumer().Consume(map[kafka.Topic]kafka.Handler{
+		&TopicQuestionAnswer{}: func(ctx context.Context, msg []byte) error {
+			data := &TopicQuestionAnswer{}
+			if err := json.Unmarshal(msg, data); err != nil {
+				return err
+			}
+			log.Infof("Kafka consume topic: %v: %#+v", data.Name(), data)
 			if data.USN == "" || data.QuestionSn == "" {
 				return utils.ErrorsWrap(errors.New("invalid data"))
 			}
 
+			// 答题记录
 			isCorrect := int32(0)
 			if data.QuestionAnswer == data.Answer {
 				isCorrect = 1
 			}
-		_, err := repository.NewRecordDao(context.Background()).Create(&model.Record{
+			_, err := repository.NewRecordDao(ctx, repository.Query()).Create(&model.Record{
 				Sn:         data.MessageSn,
 				UserSn:     data.USN,
 				QuestionSn: data.QuestionSn,
 				Answer:     data.Answer,
 				IsCorrect:  isCorrect,
 			})
+			if err != nil {
 				if errors.Is(err, gorm.ErrDuplicatedKey) {
 					return nil
 				}
 				return utils.ErrorsWrap(err)
+			}
+
+			return nil
+		},
+		&TopicAddPoint{}: func(ctx context.Context, msg []byte) error {
+			data := &TopicAddPoint{}
+			if err := json.Unmarshal(msg, data); err != nil {
+				return err
+			}
+			log.Infof("Kafka consume topic: %v: %#+v", data.Name(), data)
+			if data.USN == "" || data.Point == 0 {
+				return utils.ErrorsWrap(errors.New("invalid data"))
+			}
+
+			// 积分记录 & 加分
+			err := repository.NewPointRecordsDao(ctx, repository.Query()).CreateAndIncrPointCard(&model.PointRecord{
+				Sn:     data.MessageSn,
+				UserSn: data.USN,
+				Source: int32(data.Source),
+				Point:  data.Point,
+			})
+			if err != nil {
+				if errors.Is(err, gorm.ErrDuplicatedKey) {
+					return nil
+				}
+				return utils.ErrorsWrap(err)
+			}
+
+			return nil
+		},
 	})
 }

internal/timer/timer.go

@@ -11,7 +11,7 @@ func (t *Timer) Start() {
 	_, _ = t.c.AddFunc("0/5 * * * ?", func() {
 	})
 	t.c.Start()
-	go startConsumer()
+	startConsumer()
 }
 
 func (t *Timer) Stop() {