golang 消费kafka

使用Golang消费Kafka消息

Apache Kafka是一种高性能的分布式消息队列，被广泛应用于大数据和实时数据处理场景。在Golang中，我们可以使用第三方的kafka-go库来消费Kafka消息。

安装kafka-go

在开始之前，我们需要先安装kafka-go库。可以使用以下命令来安装kafka-go：

$ go get github.com/segmentio/kafka-go

创建Kafka消费者

首先，我们需要创建一个Kafka消费者对象。可以使用以下代码来创建消费者：

import (
    "github.com/segmentio/kafka-go"
)

func main() {
    // 设置Kafka代理地址
    brokers := []string{"localhost:9092"}

    // 创建一个新的Kafka Reader
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: brokers,
        Topic:   "my-topic",
        GroupID: "my-group",
    })

    // 循环读取消息
    for {
        // 从Kafka中读取下一条消息
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal(err)
        }

        // 处理消息
        fmt.Println(string(msg.Value))
    }
}

配置Kafka Reader

Kafka Reader配置用于指定要连接的Kafka代理地址、消费的主题和消费者组ID等信息。可以按照实际情况修改以下配置项：

Brokers: brokers,   // Kafka代理地址
Topic:   "my-topic",  // 消费的主题
GroupID: "my-group",  // 消费者组ID

消费Kafka消息

通过调用reader.ReadMessage()方法，我们可以从Kafka中读取下一条消息。

msg, err := reader.ReadMessage(context.Background())
if err != nil {
    log.Fatal(err)
}

读取的消息存储在msg变量中，可以通过msg.Value访问消息的内容。

完整示例代码

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/segmentio/kafka-go"
)

func main() {
    // 设置Kafka代理地址
    brokers := []string{"localhost:9092"}

    // 创建一个新的Kafka Reader
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: brokers,
        Topic:   "my-topic",
        GroupID: "my-group",
    })

    // 循环读取消息
    for {
        // 从Kafka中读取下一条消息
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal(err)
        }

        // 处理消息
        fmt.Println(string(msg.Value))
    }
}

总结

通过使用kafka-go库，我们可以方便地在Golang中消费Kafka消息。通过创建Kafka Reader对象，并使用ReadMessage()方法来读取消息，我们可以实现高效的Kafka消息消费功能。