feat kafka

This commit is contained in:
2026-01-24 18:12:35 +08:00
parent 6c124f9f3f
commit 69a389f32c

View File

@@ -23,7 +23,7 @@ type TopicQuestionAnswer struct {
func startConsumer() { func startConsumer() {
kafka.NewConsumer().Consume("qgdzs.question.answer", func(ctx context.Context, msg *sarama.ConsumerMessage) error { kafka.NewConsumer().Consume("qgdzs.question.answer", func(ctx context.Context, msg *sarama.ConsumerMessage) error {
log.Infof("qgdzs.question.answer: %s", string(msg.Value)) log.Infof("Kafka topic: qgdzs.question.answer: %s", string(msg.Value))
data := &TopicQuestionAnswer{} data := &TopicQuestionAnswer{}
if err := json.Unmarshal(msg.Value, &data); err != nil { if err := json.Unmarshal(msg.Value, &data); err != nil {
return utils.ErrorsWrap(err) return utils.ErrorsWrap(err)