在C#中如何确认生产者发送的消息已成功送达Kafka服务器

第一步：配置Ack参数明确确认等级 #

acks是Kafka生产者的核心配置项，用于定义服务端什么样的状态算消息接收成功，可选值如下：

• acks=0：生产者发送消息后完全不等待服务端响应，无法感知任何投递失败，不适合需要确认投递结果的场景
• acks=1：等待分区Leader节点将消息写入本地日志后就返回确认，不需要等待Follower节点同步，可靠性中等，Leader写入后立刻宕机可能丢消息
• acks=all（等价于acks=-1）：需要ISR（同步副本集合）内所有节点都完成消息写入才返回确认，可靠性最高，适合不能丢消息的业务场景

第二步：C#代码中获取投递回执的实现 #

C#生态主流使用Confluent.Kafka库操作Kafka，首先安装对应NuGet包：Confluent.Kafka。

异步发送获取确认（推荐） #

ProduceAsync方法会等待服务端按配置的acks规则返回确认，投递失败直接抛出异常，业务可以直接捕获处理：

using Confluent.Kafka;

var producerConfig = new ProducerConfig
{
    BootstrapServers = "你的Kafka节点地址:9092",
    Acks = Acks.All, // 按业务需求设置确认等级
    MessageSendMaxRetries = 3 // 可选：配置临时故障时的重试次数
};

using var producer = new ProducerBuilder<Null, string>(producerConfig).Build();
try
{
    var deliveryResult = await producer.ProduceAsync("目标Topic名称", new Message<Null, string> { Value = "待发送的消息内容" });
    // 执行到此处代表消息已按acks规则确认投递成功
    Console.WriteLine($"消息投递成功，分区：{deliveryResult.Partition}，偏移量：{deliveryResult.Offset}");
}
catch (ProduceException<Null, string> ex)
{
    // 捕获异常代表投递失败，此处执行自定义降级、重试、告警逻辑
    Console.WriteLine($"消息投递失败，错误信息：{ex.Error.Reason}");
}

回调方式获取确认 #

如果你使用无返回值的Produce方法，可以通过传入回调函数获取投递结果：

该方式适合吞吐量要求高、不需要同步等待结果的场景

producer.Produce("目标Topic名称", new Message<Null, string> { Value = "待发送的消息内容" }, deliveryReport =>
{
    if (deliveryReport.Error.IsError)
    {
        // 投递失败处理逻辑
        Console.WriteLine($"投递失败：{deliveryReport.Error.Reason}");
    }
    else
    {
        // 投递成功处理逻辑
        Console.WriteLine($"投递成功，偏移量：{deliveryReport.Offset}");
    }
});
// 退出前调用Flush确保所有未处理的投递回调都执行完成
producer.Flush(TimeSpan.FromSeconds(10));

注意事项 #

• 只要没拿到成功的返回结果或者成功的回调触发，就代表消息没有被Kafka服务端按你配置的acks规则确认接收成功
• 如果需要避免重试导致的消息重复，可以额外开启EnableIdempotence = true配置开启生产者幂等性

内容的提问来源于stack exchange，提问作者Crowley77