IceTea.Kafka 1.1.1

IceTea.Kafka (Apache Kafka 客户端类库)

简介

IceTea.Kafka 是一个基于 Confluent.Kafka 构建的 Apache Kafka 客户端类库，提供了简化的生产者和消费者API。该库封装了Kafka的核心功能，支持同步和异步消息处理，适用于构建高吞吐量的分布式消息系统。

安装

dotnet add package IceTea.Kafka

核心功能模块

一、核心类

KafkaConsumer - Kafka消息消费者

负责从Kafka主题消费消息的类。

核心属性：

• bool IsCancelled { get; set; } - 控制是否停止消费消息

核心方法：

• void Consume(string broker, string topic, string groupID, Action<ConsumerResult> action = null) - 同步消费消息
• void Consume(Func<List<ConsumerResult>, bool> func = null, Dictionary<int, int> partition = null) - 手动提交偏移量消费
• void ConsumeAsync(string broker, string topic, string groupID, Func<List<ConsumerResult>, bool> func = null) - 异步消费消息

KafkaProducer - Kafka消息生产者

负责向Kafka主题生产消息的类。

核心方法：

• void Produce(string key, string message) - 生产并发送消息

KafkaBase - Kafka基础类

提供Kafka客户端的通用功能基类。

二、配置类

ConsumerSetting - 消费者配置

Kafka消费者的配置信息。

核心属性：

• string Broker - Kafka服务器地址
• string Topic - 主题名称
• string GroupID - 消费者组ID
• Func<List<ConsumerResult>, bool> Func - 消息处理函数
• Dictionary<int, int> Partition - 分区偏移量配置

ConsumerResult - 消费结果

封装消费到的消息信息。

核心属性：

• string Broker - Kafka服务器地址
• string Topic - 主题名称
• Partition Partition - 分区信息
• long Offset - 消息偏移量
• string Message - 消息内容

使用指南

1. 消费者使用示例

// 创建消费者配置
var consumerSetting = new ConsumerSetting
{
    Broker = "localhost:9092",
    Topic = "test-topic",
    GroupID = "test-group"
};

// 创建消费者实例
var consumer = new KafkaConsumer(consumerSetting);

// 同步消费消息
consumer.Consume("localhost:9092", "test-topic", "test-group", result =>
{
    Console.WriteLine($"收到消息: {result.Message}");
    Console.WriteLine($"主题: {result.Topic}, 分区: {result.Partition}");
});

// 异步消费消息
consumer.ConsumeAsync("localhost:9092", "test-topic", "test-group", results =>
{
    foreach (var result in results)
    {
        Console.WriteLine($"处理消息: {result.Message}");
    }
    return true; // 确认消息处理完成
});

2. 生产者使用示例

// 创建生产者配置
var producerSetting = new ConsumerSetting
{
    Broker = "localhost:9092",
    Topic = "test-topic"
};

// 创建生产者实例
var producer = new KafkaProducer(producerSetting);

// 发送消息
producer.Produce("key1", "Hello Kafka!");
producer.Produce("key2", "这是一条测试消息");

3. 手动提交偏移量

var consumerSetting = new ConsumerSetting
{
    Broker = "localhost:9092",
    Topic = "test-topic",
    GroupID = "manual-commit-group"
};

var consumer = new KafkaConsumer(consumerSetting);

// 指定分区和起始偏移量
var partitionOffsets = new Dictionary<int, int>
{
    { 0, 100 }, // 分区0从偏移量100开始
    { 1, 200 }  // 分区1从偏移量200开始
};

// 手动提交偏移量消费
consumer.Consume(results =>
{
    foreach (var result in results)
    {
        // 处理消息
        ProcessMessage(result.Message);
    }
    return true; // 返回true表示确认处理完成，可以提交偏移量
}, partitionOffsets);

4. 停止消费

// 停止消费循环
consumer.IsCancelled = true;

技术特性

消费模式

• 自动提交：由Kafka自动管理偏移量提交
• 手动提交：应用程序控制偏移量提交时机
• 同步消费：阻塞式消息处理
• 异步消费：非阻塞式消息处理

配置选项

• BootstrapServers：Kafka集群地址
• GroupId：消费者组标识
• AutoOffsetReset：偏移量重置策略
• EnableAutoCommit：自动提交开关
• SessionTimeoutMs：会话超时时间

错误处理

• 自动重连机制
• 异常安全处理
• 详细的日志记录
• 消费失败恢复

典型应用场景

1. 微服务通信：服务间异步消息传递
2. 日志收集：集中式日志处理系统
3. 事件驱动架构：基于事件的业务流程
4. 数据管道：实时数据流处理
5. 消息队列：可靠的消息传输

最佳实践

1. 消费者组管理

// 为不同的业务逻辑使用不同的消费者组
var orderConsumer = new KafkaConsumer(new ConsumerSetting 
{ 
    GroupID = "order-processing-group" 
});

var notificationConsumer = new KafkaConsumer(new ConsumerSetting 
{ 
    GroupID = "notification-group" 
});

2. 错误处理

try
{
    consumer.Consume(broker, topic, groupId, result =>
    {
        try
        {
            ProcessMessage(result.Message);
        }
        catch (Exception ex)
        {
            // 记录处理错误，但不中断消费
            logger.Error($"消息处理失败: {ex.Message}", ex);
        }
    });
}
catch (Exception ex)
{
    logger.Fatal("消费者启动失败", ex);
}

3. 资源管理

// 使用using确保资源正确释放
using (var consumer = new KafkaConsumer(setting))
{
    consumer.Consume(broker, topic, groupId, ProcessMessage);
}

依赖说明

• Confluent.Kafka: 核心Kafka客户端库
• IceTea.Atom: 基础工具类库
• IceTea.Pure: 核心基础库

兼容性

• Kafka版本: 0.10.0+
• .NET版本: .NET Standard 2.0+
• 平台: Windows, Linux, macOS

许可证

MIT License

作者

WuMing

贡献

欢迎提交 Issue 和 Pull Request！