package timer import ( "context" "errors" "git.hlsq.asia/alex/service-common/db/kafka" "git.hlsq.asia/alex/service-common/log" "git.hlsq.asia/alex/service-common/utils" "git.hlsq.asia/alex/service-qgdzs/internal/dao/model" "git.hlsq.asia/alex/service-qgdzs/internal/dao/repository" "gorm.io/gorm" ) // !!!消费者必须做幂等!!! func startConsumer() { go kafka.NewConsumer().Consume([]kafka.Topic{ &TopicQuestionAnswer{}, &TopicAddPoint{}, }) } type TopicQuestionAnswer struct { kafka.TopicQuestionAnswer } func (t *TopicQuestionAnswer) OnMessage(ctx context.Context) error { log.Infof("Kafka consume topic: %v: %#+v", t.Name(), t) if t.USN == 0 || t.QuestionSn == 0 { return utils.ErrorsWrap(errors.New("invalid data")) } // 答题记录 isCorrect := int32(0) if t.QuestionAnswer == t.Answer { isCorrect = 1 } _, err := repository.NewRecordDao(ctx, repository.Query()).Create(&model.Record{ Sn: t.MessageSn, UserSn: t.USN, QuestionSn: t.QuestionSn, Answer: t.Answer, IsCorrect: isCorrect, }) if err != nil { if errors.Is(err, gorm.ErrDuplicatedKey) { return nil } return utils.ErrorsWrap(err) } return nil } type TopicAddPoint struct { kafka.TopicAddPoint } func (t *TopicAddPoint) OnMessage(ctx context.Context) error { log.Infof("Kafka consume topic: %v: %#+v", t.Name(), t) if t.USN == 0 || t.Point == 0 { return utils.ErrorsWrap(errors.New("invalid data")) } // 积分记录 & 加分 err := repository.NewPointRecordsDao(ctx, repository.Query()).CreateAndIncrPointCard(&model.PointRecord{ Sn: t.MessageSn, UserSn: t.USN, Source: int32(t.Source), Point: t.Point, }) if err != nil { if errors.Is(err, gorm.ErrDuplicatedKey) { return nil } return utils.ErrorsWrap(err) } return nil }