Go语言实时数据处理：Sarama Kafka

在Go生态中，处理Kafka消息时，Sarama是一个广泛使用的库。它功能扎实、API设计顺手，非常适合实时数据处理场景。下面将逐步介绍如何使用Sarama进行Kafka消息的消费。

安装Sarama库

首先，需要安装Sarama库。在终端中执行以下命令：

go get github.com/Shopify/sarama

编写消费者示例

安装完成后，可以编写一个消费者程序。以下示例演示了如何创建一个消费者组，并订阅名为test_topic的主题。消费者组名为test_group，使用exampleConsumerGroupHandler处理消息。当消息到达时，会直接打印到控制台。

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "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 := "test_topic"

    consumer, err := sarama.NewConsumerGroup(brokers, sarama.ConsumerGroupConfig{
        ClientID:    "test_group",
        GroupVersion: config.Version,
        Config:      config,
    })
    if err != nil {
        log.Fatalf("Error creating consumer group client: %v", err)
    }
    defer func() {
        if err := consumer.Close(); err != nil {
            log.Fatalf("Error closing consumer: %v", err)
        }
    }()

    handler := exampleConsumerGroupHandler{}

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

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

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() {
        fmt.Printf("Message: %s\n", string(msg.Value))
        sess.MarkMessage(msg, "")
    }
    return nil
}

运行程序

代码编写完成后，使用go run main.go命令运行即可。只要向test_topic发送消息，程序就会实时捕获并打印到控制台。如果需要进一步处理数据，可以在ConsumeClaim方法中扩展逻辑，例如解析、过滤或存储到数据库。