Kafka Konsumer 使用指南

Kafka Konsumer 是一个简化 Kafka 消费者实现的 Go 语言库，集成了内置异常管理机制（kafka-cronsumer），旨在提供更便捷、健壮的消息处理能力。以下是对该开源项目的快速入门教程，包括项目结构、启动文件和配置文件的简介。

1. 项目目录结构及介绍

本项目遵循 Go 的标准工作区结构：

kafka-konsumer/
├── api                 # API 相关代码
├── examples            # 示例应用
│   ├── ...             # 各种使用场景的示例代码
├── goreleaser.yml      # 自动发布配置文件
├── golangci.yml        # Golang CI 配置
├── go.mod              # Go 模块依赖声明
├── go.sum              # 依赖校验文件
├── LICENSE             # 许可证文件
├── Makefile            # 构建和测试任务的脚本
├── README.md           # 项目读我文件
├── SECURITY.md         # 安全相关说明
└── src                 # 主要业务逻辑代码
    ├── collector       # 消息收集器相关
    ├── consumer        # 消费者核心逻辑
    ├── data_units      # 数据单元处理
    ├── ...             # 其它如生产者、层管理、日志处理等代码模块

• src 目录包含了主要的业务逻辑，包括消费者 (consumer), 生产者 (producer), 以及各种辅助组件。
• examples 提供了多个示例程序，涵盖了基础消费、带重试与异常处理的消费等场景。
• configurations 并未作为一个单独的目录列出，但配置通常通过在代码中定义的结构体来实现，例如 kafka.ConsumerConfig。

2. 项目的启动文件介绍

在 Kafka Konsumer 中，并没有一个明确的“启动文件”作为传统意义上的入口点。然而，开发者应当从 main() 函数开始他们的应用，这个函数位于示例代码或自定义应用程序中。以简单的消费为例，可以从 examples 目录下的某个示例开始，比如一个基本的消费者示例可能看起来像这样：

package main

import (
    "fmt"
    "github.com/Trendyol/kafka-konsumer/v2"
)

func consumeFn(message kafka.Message) error {
    fmt.Printf("Message From %s with value %s\n", message.Topic, string(message.Value))
    return nil
}

func main() {
    consumerCfg := &kafka.ConsumerConfig{
        Reader: kafka.ReaderConfig{
            Brokers: []string{"localhost:29092"},
            Topic:   "your-topic",
            GroupID: "group-id",
        },
        ConsumeFn: consumeFn,
    }
    
    consumer, _ := kafka.NewConsumer(consumerCfg)
    defer consumer.Stop()
    consumer.Consume()
}

在这个例子中，main() 函数初始化了一个消费者配置，并调用了 kafka.NewConsumer 来创建消费者实例，然后开始消费过程。

3. 项目的配置文件介绍

Kafka Konsumer 的配置是通过代码中的结构体实例来设定的，而不是通过外部的配置文件加载。这意味着配置项直接在 Go 代码内被定义，例如：

• ConsumerConfig: 包含了消费者的基本设置，如 ReaderConfig（定义了 Kafka 服务器地址、主题和组ID）。
• ReaderConfig: 控制如何连接到 Kafka，如 Broker 列表、主题等。
• RetryConfiguration: 当启用重试时的详细配置，如重试主题、重试时间间隔和最大重试次数。

开发者应直接在代码中按需修改这些配置项。例如，如果你想要开启消息重试，你需要设置 RetryEnabled 为 true 并定义相应的 RetryConfiguration。

示例配置片段

consumerCfg := &kafka.ConsumerConfig{
    Reader: kafka.ReaderConfig{
        Brokers: []string{"localhost:29092"}, // Kafka集群地址
        Topic:   "standard-topic",            // 要监听的主题
        GroupID: "example-group",             // 消费者组ID
    },
    RetryEnabled: true,                      // 启用重试功能
    RetryConfiguration: kafka.RetryConfiguration{
        Topic:          "retry-topic",       // 重试消息发送到的主题
        StartTimeCron:  "*/1 * * * *",        // 重试策略启动时间（cron表达式）
        WorkDuration:   time.Minute,          // 工作持续时间
        MaxRetry:       5,                   // 最大重试次数
    },
    ConsumeFn: myConsumeFunction,             // 消费消息的回调函数
}

请注意，实际应用中，根据具体需求调整以上配置参数，并替换示例中的占位符（如主题名）。这种方式确保了配置的高度定制性，但也要求开发者在编译时决定配置细节。