Confluent.Kafka 是 C# 生态中事实标准的 Kafka 客户端，安装需通过 NuGet 引入最新稳定版（如 2.4.0），推荐用 docker-compose 快速启动本地 Kafka（含 ZooKeeper），生产者须设 enable.idempotence=true、acks=all、retries=int.MaxValue，消费者应固定 GroupId、手动提交 offset 并明确 AutoOffsetReset 策略。
Confluent.Kafka 是当前 C# 生态中事实标准的 Kafka 客户端，稳定、性能好、文档全、社区活跃。别再用已归档的 KafkaNet 或维护停滞的 rdkafka-dotnet —— 后者虽曾提供 EventConsumer 这类便利封装，但底层绑定的是旧版 librdkafka，.NET 6+ 下容易出兼容问题，且不支持 group.instance.id 等关键消费者语义。
怎么装包和连上 Kafka（Windows / Docker 最简验证路径）
本地开发验证，推荐用 docker-compose 起单节点 Kafka + ZooKeeper（注意：ZooKeeper 已在 Kafka 3.3+ 中废弃，但 Confluent.Kafka 当前仍兼容旧协议，且本地调试用它最省事）。
新建 docker-compose.yml，内容如下：
version: "3"
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.3.2
environment:
ZOOKEEPER_CLIENT_PORT: 2181
kafka:
image: confluentinc/cp-kafka:7.3.2
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1执行 docker-compose up -d 启动后，用以下命令快速建 Topic 验证连通性：
.\bin\windows\kafka-topics.bat --bootstrap-server localhost:9092 --create --topic test --partitions 1 --replication-factor 1
在 C# 项目中安装 NuGet 包：Confluent.Kafka（最新稳定版，如 2.4.0），不要装 Confluent.Kafka.SchemaRegistry 等扩展包，除非你真要用 Avro。
生产者怎么发消息不丢（acks / retries / idempotence 必设）
默认配置下，Producer 可能静默丢消息——比如网络抖动时重试未启用，或 broker 返回失败但客户端没感知。
关键配置必须显式设置：
-
bootstrap.servers：填"localhost:9092"，多个用逗号分隔；不能写127.0.0.1，否则容器内 DNS 解析失败 -
enable.idempotence：设为true，这是“至少一次”投递的底线，要求max.in.flight.requests.per.connection （默认满足） -
acks：设为"all"，确保 ISR（同步副本）全部写入才返回成功 -
retries：设为int.MaxValue（或一个大值如10），配合retry.backoff.ms（默认 100ms） -
key.serializer和value.serializer：必须指定，哪怕只是System.Text.UTF8Encoding，否则运行时报NullReferenceException
示例初始化：
var config = new ProducerConfig
{
BootstrapServers = "localhost:9092",
EnableIdempotence = true,
Acks = Acks.All,
Retries = int.MaxValue,
KeySerializer = new StringSerializer(Encoding.UTF8),
ValueSerializer = new StringSerializer(Encoding.UTF8)
};消费者怎么避免重复消费或跳过消息（GroupId / AutoOffsetReset / Commit）
重复消费 ≠ Bug，而是 Kafka 的设计特性；真正要防的是“本该处理却漏了”。核心在三处：
-
GroupId必须固定且有意义，例如"order-processor-v2"，不能每次 new 一个 Guid -
AutoOffsetReset别依赖默认值：首次启动想读历史数据就设AutoOffsetReset.Earliest；只想读新消息就设AutoOffsetReset.Latest - 手动提交（
consumer.Commit()）比自动提交更可控，尤其在业务逻辑含 DB 写入时——应“先处理完业务，再 Commit offset”，否则 crash 后会重复处理
注意：如果用了 group.instance.id（Kafka 3.3+ 支持），消费者重启不会触发再均衡，offset 恢复更快，但需确保同一时刻只有一个实例在跑，否则报错 GROUP_INSTANCE_ID_CONFLICT。
基础消费循环结构：
using var consumer = new ConsumerBuilder<Ignore, string>(config).Build();
consumer.Subscribe("test");
try
{
while (true)
{
var consumeResult = consumer.Consume(cts.Token);
ProcessMessage(consumeResult.Message.Value);
consumer.Commit(consumeResult); // 手动提交
}
}
catch (OperationCanceledException) { }序列化器自己写要注意什么（byte[] / UTF8 / Null 值）
别直接用 JsonSerializer<t></t>（来自 Confluent.Kafka.Serialization.Json），它内部没做 null 安全，遇到 null 值会抛 ArgumentNullException。
更稳妥的做法是：
- Key 用
StringSerializer或自定义NullSafeStringSerializer（对 null 返回空字节数组） - Value 若是 JSON，用
System.Text.Json自封装，捕获JsonSerializer.SerializeToUtf8Bytes异常并 fallback - 所有序列化器必须线程安全：不能在内部缓存可变状态，
Serialize方法不能有副作用
一个最小可用的 null-safe string 序列化器：
public class NullSafeStringSerializer : ISerializer<string>
{
public byte[] Serialize(string data, SerializationContext context)
=> data == null ? Array.Empty<byte>() : Encoding.UTF8.GetBytes(data);
}真正麻烦的从来不是“怎么连上”，而是“怎么保证每条消息都按预期被处理一次”。enable.idempotence 和 acks=all 解决生产端可靠性，GroupId + manual commit + 明确的 AutoOffsetReset 策略解决消费端语义。其他配置如 batch.size、compression.type 属于调优项，初期不用碰。别在没搞清 offset 提交时机时就开自动提交，那是重复消费的头号来源。
