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变量。