在Golang中，我们可以使用sarama库来操作Kafka并实现消息过滤。以下是一个简单的示例，展示了如何使用sarama库创建一个消费者，过滤特定主题的消息，并将过滤后的消息发送到另一个主题。

首先，确保已经安装了sarama库：

go get github.com/Shopify/sarama

然后，编写以下代码：

package mainimport ("fmt""log""os""os/signal""strings""sync""github.com/Shopify/sarama")func main() {config := sarama.NewConfig()config.Consumer.Return.Errors = trueconfig.Version = sarama.V2_6_0_0brokers := []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)<-signalshandler.wg.Wait()consumer.Close()}type exampleConsumerGroupHandler struct {consumer sarama.ConsumerGrouptopicFilter[]stringfilteredTopic stringwg 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变量。