在 C# (.NET) 中使用 Kafka | Confluent.Kafka

用于在 C# (.NET) 中操作 Apache Kafka 的 Confluent.Kafka 库

Apache Kafka 是一个分布式流数据处理平台，已成为构建事件驱动架构和微服务的标准。对于 C# (.NET) 开发者来说，最流行且受支持的库是 Confluent.Kafka——由 Confluent 公司提供的官方客户端，基于高性能 C 客户端 librdkafka 构建。在本文中，我们将详细介绍如何安装该库、创建生产者和消费者，并探讨最佳实践。

安装 Confluent.Kafka

首先，通过 NuGet 将包添加到项目中：

dotnet add package Confluent.Kafka

或者通过包管理器控制台：

Install-Package Confluent.Kafka

生产者 (Producer) 的基本配置

生产者将消息发送到 Kafka 主题。来看一个简单示例：

using Confluent.Kafka;

var config = new ProducerConfig{    BootstrapServers = "localhost:9092"};

using (var producer = new ProducerBuilder<string, string>(config).Build()){    var message = new Message<string, string>    {        Key = "user_123",        Value = "你好，Kafka！"    };

    var deliveryReport = await producer.ProduceAsync("my-topic", message);    Console.WriteLine($"消息已发送。分区: {deliveryReport.Partition}, 偏移量: {deliveryReport.Offset}");}

关键点：

BootstrapServers — 一个或多个 Kafka 代理的地址。
ProduceAsync — 异步发送并等待确认。
Key 可以保证同一键下消息的顺序性。

使用序列化处理复杂对象的生产者

通常需要发送 JSON 或二进制数据，而非字符串。使用自定义序列化器或 Newtonsoft.Json 库：

using System.Text.Json;using Confluent.Kafka;

var config = new ProducerConfig { BootstrapServers = "localhost:9092" };

var producer = new ProducerBuilder<string, Order>(config)    .SetValueSerializer(new JsonSerializer<Order>())    .Build();

var order = new Order { Id = 1, Amount = 150.50m, Customer = "伊万" };await producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });

public class Order{    public int Id { get; set; }    public decimal Amount { get; set; }    public string Customer { get; set; }}

public class JsonSerializer<T> : ISerializer<T>{    public byte[] Serialize(T data, SerializationContext context) =>        JsonSerializer.SerializeToUtf8Bytes(data);}

消费者 (Consumer) 的配置

消费者从一个或多个主题中读取消息。订阅与处理示例：

using Confluent.Kafka;

var config = new ConsumerConfig{    BootstrapServers = "localhost:9092",    GroupId = "my-consumer-group",    AutoOffsetReset = AutoOffsetReset.Earliest};

using (var consumer = new ConsumerBuilder<string, string>(config).Build()){    consumer.Subscribe("my-topic");

    CancellationTokenSource cts = new CancellationTokenSource();    Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

    try    {        while (true)        {            var consumeResult = consumer.Consume(cts.Token);            Console.WriteLine($"已接收: Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");        }    }    catch (OperationCanceledException)    {        consumer.Close();    }}

重要配置参数：