Go语言使用kafka-go实现Kafka消费消息_Golang

在这篇教程中，我们将介绍如何使用 kafka-go 库来消费 kafka 消息，并重点讲解 fetchmessage 和 readmessage 的区别，以及它们各自适用的场景。通过这篇教程，你将了解如何有效地使用 kafka-go 库来处理消息和管理偏移量。

安装 kafka-go 库

首先，你需要在项目中安装 kafka-go 库。可以使用以下命令：

go get github.com/segmentio/kafka-go

初始化 kafka reader

为了从 kafka 消费消息，我们首先需要配置和初始化 kafka reader。以下是一个简单的 kafka reader 初始化示例：

package main

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

func main() {
    // 创建 kafka reader
    kafkareader := kafka.newreader(kafka.readerconfig{
        brokers:   []string{"localhost:9092"}, // kafka broker 地址
        topic:     "example-topic",            // 订阅的 kafka topic
        groupid:   "example-group",            // 消费者组 id
        partition: 0,                          // 分区号 (可选)
        minbytes:  10e3,                       // 10kb
        maxbytes:  10e6,                       // 10mb
    })

    defer kafkareader.close()
}

使用 fetchmessage 消费消息

fetchmessage 允许你从 kafka 消费消息并手动提交偏移量，这给你对消息处理的更精确控制。以下是如何使用 fetchmessage 的示例：

func consumewithfetchmessage() {
    ctx := context.background()
    
    for {
        // 从 kafka 中获取下一条消息
        m, err := kafkareader.fetchmessage(ctx)
        if err != nil {
            log.printf("获取消息时出错: %v", err)
            break
        }

        // 打印消息内容
        log.printf("消息: %s, 偏移量: %d", string(m.value), m.offset)

        // 处理消息 (在这里可以进行你的业务逻辑)

        // 手动提交偏移量
        if err := kafkareader.commitmessages(ctx, m); err != nil {
            log.printf("提交偏移量时出错: %v", err)
        }
    }
}

优点

精确控制偏移量：在处理消息后，你可以手动选择是否提交偏移量，这样可以确保只有在消息处理成功后才提交。
重试机制：可以灵活地处理失败消息，例如在处理失败时，不提交偏移量，从而实现消息的重新消费。

缺点

代码复杂度增加：需要手动处理偏移量提交，会增加一些额外的代码量。

使用 readmessage 消费消息

readmessage 是一种更简单的方式，从 kafka 中获取消息并自动提交偏移量。适用于对消费逻辑不太敏感的场景。以下是使用 readmessage 的示例：

func consumewithreadmessage() {
    ctx := context.background()
    
    for {
        // 从 kafka 中读取下一条消息并自动提交偏移量
        datainfo, err := kafkareader.readmessage(ctx)
        if err != nil {
            log.printf("读取消息时出错: %v", err)
            break
        }

        // 打印消息内容
        log.printf("消息: %s, 偏移量: %d", string(datainfo.value), datainfo.offset)

        // 处理消息 (在这里可以进行你的业务逻辑)
    }
}

优点

简单易用：readmessage 自动提交偏移量，代码简洁，易于维护。
快速开发：适合简单的消息处理逻辑和对消息可靠性要求不高的场景。

缺点

缺乏灵活性：无法在处理失败时重新消费消息，因为偏移量已经自动提交。

总结选择

方法	优点	缺点	适用场景
`fetchmessage`	需要手动提交偏移量，精确控制消息处理和提交逻辑	代码复杂度较高	需要精确控制消息处理的场景，例如处理失败重试
`readmessage`	简单易用，自动提交偏移量，代码更简洁	无法重新消费已处理失败的消息	简单的消息处理，对消息处理成功率要求不高的场景

完整示例

以下是一个完整的 kafka 消费者示例，包括 fetchmessage 和 readmessage 两种方法。可以根据你的需求选择合适的方法：

package main

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

func main() {
    // 创建 kafka reader
    kafkareader := kafka.newreader(kafka.readerconfig{
        brokers:   []string{"localhost:9092"},
        topic:     "example-topic",
        groupid:   "example-group",
        minbytes:  10e3, // 10kb
        maxbytes:  10e6, // 10mb
    })

    defer kafkareader.close()

    // 使用 fetchmessage 消费消息
    log.println("开始使用 fetchmessage 消费 kafka 消息...")
    consumewithfetchmessage(kafkareader)

    // 使用 readmessage 消费消息
    log.println("开始使用 readmessage 消费 kafka 消息...")
    consumewithreadmessage(kafkareader)
}

func consumewithfetchmessage(kafkareader *kafka.reader) {
    ctx := context.background()

    for {
        m, err := kafkareader.fetchmessage(ctx)
        if err != nil {
            log.printf("fetchmessage 获取消息时出错: %v", err)
            break
        }

        log.printf("fetchmessage 消息: %s, 偏移量: %d", string(m.value), m.offset)

        // 手动提交偏移量
        if err := kafkareader.commitmessages(ctx, m); err != nil {
            log.printf("fetchmessage 提交偏移量时出错: %v", err)
        }
    }
}

func consumewithreadmessage(kafkareader *kafka.reader) {
    ctx := context.background()

    for {
        datainfo, err := kafkareader.readmessage(ctx)
        if err != nil {
            log.printf("readmessage 读取消息时出错: %v", err)
            break
        }

        log.printf("readmessage 消息: %s, 偏移量: %d", string(datainfo.value), datainfo.offset)
    }
}