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

go get github.com/Shopify/sarama

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

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

package mainimport ("fmt""log""os""github.com/Shopify/sarama")func main() {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.WaitForAllconfig.Producer.Retry.Max = 5config.Producer.Return.Successes = trueproducer, 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)}

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

package mainimport ("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 = trueconfig.Version = sarama.V2_6_0_0consumer, 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 MyMessageerr := 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的主题。