golang使用kafka如何进行消息解码

在Golang中，我们可以使用sarama库来处理Kafka消息的编码和解码。首先，确保已经安装了sarama库：

go get github.com/Shopify/sarama

接下来，我们将创建一个简单的生产者，将消息发送到Kafka，然后创建一个消费者，从Kafka接收消息并解码它们。

1. 创建一个生产者，将消息发送到Kafka：

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/Shopify/sarama"
)

func main() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true

	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
	if err != nil {
		log.Fatalln("错误: 创建生产者失败:", err)
	}
	defer func() {
		if err := producer.Close(); err != nil {
			log.Fatalln("错误: 关闭生产者失败:", err)
		}
	}()

	msg := &sarama.ProducerMessage{
		Topic: "test_topic",
		Value: sarama.StringEncoder("Hello, Kafka!"),
	}

	partition, offset, err := producer.SendMessage(msg)
	if err != nil {
		log.Printf("发送消息失败: %v\n", err)
		return
	}

	fmt.Printf("消息已发送至分区 %d 偏移量 %d\n", partition, offset)
}

2. 创建一个消费者，从Kafka接收消息并解码它们：

package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"log"
	"os"

	"github.com/Shopify/sarama"
)

type MyMessage struct {
	Content string `json:"content"`
}

func main() {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V2_6_0_0

	consumer, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "test_group", config)
	if err != nil {
		log.Fatalln("错误: 创建消费者组失败:", err)
	}
	defer func() {
		if err := consumer.Close(); err != nil {
			log.Fatalln("错误: 关闭消费者组失败:", err)
		}
	}()

	handler := exampleConsumerGroupHandler{}
	err = consumer.Consume(context.Background(), []string{"test_topic"}, handler)
	if err != nil {
		log.Fatalln("错误: 消费消息失败:", err)
	}
}

type exampleConsumerGroupHandler struct{}

func (exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		var myMsg MyMessage
		err := json.Unmarshal(msg.Value, &myMsg)
		if err != nil {
			fmt.Printf("错误: 解码消息失败: %v\n", err)
			continue
		}
		fmt.Printf("解码后的消息: %+v\n", myMsg)

		sess.MarkMessage(msg, "")
	}

	return nil
}

在这个例子中，我们创建了一个名为MyMessage的结构体，用于存储解码后的消息。ConsumeClaim方法中的json.Unmarshal函数用于将接收到的消息解码为MyMessage结构体。请确保在运行消费者之前，Kafka中已经存在名为test_topic的主题。