为什么不建议用Kafka实现请求响应模式？.NET生态下有何替代方案？

业内的反对意见本质是源于Kafka的底层设计定位和请求响应场景的需求不匹配，核心原因包括：

• Kafka的设计目标是高吞吐的流式异步传输、日志持久化，原生没有RPC场景需要的超时、重试、请求-响应绑定、故障容错的语义支持，所有能力都需要上层业务自行封装，重复造轮子的成本极高
• 消费模型是批量拉取模式，单请求的端到端延迟抖动远高于专门的RPC框架或者同步消息中间件：完整链路需要经过「请求生产→消费者拉取请求→处理→响应生产→请求端拉取响应」至少4次网络IO+2次磁盘持久化（按默认配置），阻塞等待HTTP响应的场景下非常容易触发前端超时
• 资源利用率和可用性差：每个请求端服务实例都需要消费响应Topic，服务扩缩容会触发消费者组重平衡，重平衡期间整个响应消费会停滞，进一步加剧超时问题；如果请求端发送请求后宕机，对应的响应消息会成为无效死消息，需要额外的过期清理逻辑
• 容错成本高：需要自行处理请求丢失、重复消费、部分失败、幂等校验的问题，没有成熟的开箱即用实现，线上出问题排查难度极高

Spring Kafka提供的请求响应实现只是上层封装了correlationId映射、响应消费、超时控制的逻辑，并没有解决Kafka底层的模型缺陷，仅适合非核心的低并发同步场景，不推荐用在核心链路。

方案1：硬基于Kafka封装（仅建议测试、非核心场景使用） #

Confluent.Kafka和MassTransit没有提供开箱即用的实现，是因为官方团队认为Kafka不适合这类场景，如果你暂时只能用Kafka，可以自行封装核心逻辑：

1. 实现全局单例的响应Topic消费者，后台持续拉取响应消息
2. 维护ConcurrentDictionary<string, TaskCompletionSource<string>>存储待完成的请求，key为correlationId，value为等待结果的TaskCompletionSource
3. 发送请求时将correlationId和对应的TaskCompletionSource存入字典，同时设置超时自动移除并标记任务取消
4. 响应消费者拿到消息后，根据correlationId匹配对应的TaskCompletionSource，设置返回结果
   核心代码示例如下：

public class KafkaRequestResponseClient : IDisposable
{
    private readonly IProducer<string, string> _producer;
    private readonly IConsumer<string, string> _responseConsumer;
    private readonly ConcurrentDictionary<string, TaskCompletionSource<string>> _pendingRequests = new();
    private readonly CancellationTokenSource _cts = new();

    public KafkaRequestResponseClient(KafkaConfig config)
    {
        _producer = new ProducerBuilder<string, string>(new ProducerConfig { BootstrapServers = config.BootstrapServers }).Build();
        _responseConsumer = new ConsumerBuilder<string, string>(new ConsumerConfig
        {
            BootstrapServers = config.BootstrapServers,
            GroupId = config.ResponseConsumerGroupId,
            AutoOffsetReset = AutoOffsetReset.Latest
        }).Build();
        _responseConsumer.Subscribe(config.ResponseTopics);
        // 启动后台消费线程
        _ = Task.Run(StartConsuming, _cts.Token);
    }

    private async Task StartConsuming()
    {
        while (!_cts.IsCancellationRequested)
        {
            var result = _responseConsumer.Consume(_cts.Token);
            var correlationId = Encoding.UTF8.GetString(result.Message.Headers.First(h => h.Key == "CorrelationId").GetValueBytes());
            if (_pendingRequests.TryRemove(correlationId, out var tcs))
            {
                tcs.TrySetResult(result.Message.Value);
            }
        }
    }

    public Task<string> SendRequest(string requestTopic, string payload, string correlationId, TimeSpan? timeout = null)
    {
        timeout ??= TimeSpan.FromSeconds(15);
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        if (!_pendingRequests.TryAdd(correlationId, tcs))
        {
            throw new InvalidOperationException("重复的请求ID");
        }
        // 发送请求消息
        _producer.Produce(requestTopic, new Message<string, string>
        {
            Value = payload,
            Headers = new Headers { { "CorrelationId", Encoding.UTF8.GetBytes(correlationId) } }
        });
        // 超时控制
        _ = Task.Delay(timeout.Value, _cts.Token).ContinueWith(_ =>
        {
            if (_pendingRequests.TryRemove(correlationId, out _))
            {
                tcs.TrySetException(new TimeoutException("请求超时"));
            }
        }, _cts.Token);
        return tcs.Task;
    }

    public void Dispose()
    {
        _cts.Cancel();
        _producer.Dispose();
        _responseConsumer.Dispose();
    }
}

你的业务代码只要注入上面的客户端，把RequetOne和RequetSecond改成调用SendRequest方法即可，无需修改现有聚合逻辑。

方案2：生产环境标准方案（推荐） #

根据你的余额聚合查询场景，优先选择更适配的技术栈：

• 优先选择gRPC：如果三个余额查询服务都是内部微服务，.NET原生支持gRPC，性能高、延迟低，原生支持超时、重试、取消令牌，完全适配你现有的并行聚合代码逻辑，是同步查询场景的首选
• 如果需要消息解耦，换用RabbitMQ：RabbitMQ原生支持临时回复队列、请求响应语义，.NET下MassTransit已经封装了开箱即用的IRequestClient，你无需自行维护correlationId和任务映射，直接调用即可拿到返回结果
• 如果可以接受异步交互：调整前端交互逻辑，发送请求后立即返回查询ID，前端轮询或者通过WebSocket推送结果，这时候用Kafka是合理的，完全发挥Kafka高吞吐、高可用的优势，也不会有HTTP请求阻塞的问题

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