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的主題。