在Golang中，我们可以使用sarama库来处理Kafka消息的编码和解码。首先，确保已经安装了sarama库：
云服务器
物理服务器
虚拟主机
跨境电商服务器go get github.com/Shopify/sarama
接下来，我们将创建一个简单的生产者，将消息发送到Kafka，然后创建一个消费者，从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)
}
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的主题。
辰迅云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
推荐阅读： kafka集群部署与其他技术栈如何集成