golang操作kafka如何實現消息過濾

在Golang中，我們可以使用sarama庫來操作Kafka并實現消息過濾。以下是一個簡單的示例，展示了如何使用sarama庫創建一個消費者，過濾特定主題的消息，并將過濾后的消息發送到另一個主題。

首先，確保已經安裝了sarama庫：

go get github.com/Shopify/sarama

然后，編寫以下代碼：

package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"sync"

	"github.com/Shopify/sarama"
)

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

	brokers := []string{"localhost:9092"}
	topic := "filter-topic"
	filteredTopic := "filtered-topic"

	consumer, err := sarama.NewConsumerGroup(brokers, config)
	if err != nil {
		log.Fatalf("Error creating consumer group: %v", err)
	}
	defer consumer.Close()

	topicFilter := strings.Split(topic, ",")

	handler := exampleConsumerGroupHandler{
		consumer:   consumer,
		topicFilter: topicFilter,
		filteredTopic: filteredTopic,
		wg: sync.WaitGroup{},
	}

	err = consumer.Consume(context.Background(), []string{topic}, &handler)
	if err != nil {
		log.Fatalf("Error consuming messages: %v", err)
	}

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	<-signals

	handler.wg.Wait()
	consumer.Close()
}

type exampleConsumerGroupHandler struct {
	consumer     sarama.ConsumerGroup
	topicFilter  []string
	filteredTopic string
	wg           sync.WaitGroup
}

func (h *exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (h *exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

func (h *exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		for _, t := range h.topicFilter {
			if strings.HasPrefix(msg.Topic, t) {
				h.wg.Add(1)
				go func(msg *sarama.ConsumerMessage) {
					defer h.wg.Done()
					h.sendToFilteredTopic(msg)
				}(msg)
				break
			}
		}
	}

	return nil
}

func (h *exampleConsumerGroupHandler) sendToFilteredTopic(msg *sarama.ConsumerMessage) {
	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, h.consumer.Config())
	if err != nil {
		log.Printf("Error creating producer: %v", err)
		return
	}
	defer producer.Close()

	msg.Key = sarama.ByteEncoder(msg.Key)
	msg.Value = sarama.ByteEncoder(msg.Value)

	partition, offset, err := sess.DefaultPartitionInfo(msg.Topic)
	if err != nil {
		log.Printf("Error getting partition info: %v", err)
		return
	}

	_, _, err = producer.SendMessage(&sarama.ProducerMessage{
		Topic:     h.filteredTopic,
		Partition: partition,
		Offset:    offset,
		Key:       msg.Key,
		Value:     msg.Value,
	})

	if err != nil {
		log.Printf("Error sending message to filtered topic: %v", err)
	}
}

這個示例中，我們創建了一個消費者組，訂閱了filter-topic主題。消費者組的處理器exampleConsumerGroupHandler會檢查每條消息的主題是否包含在topicFilter中。如果包含，則將消息發送到filtered-topic主題。

注意：這個示例假設你已經有一個運行在localhost:9092的Kafka集群。你需要根據實際情況修改brokers變量。