如何基于Kafka实现API网关与微服务的HTTP同步请求响应？

问题描述 #

我正在开发一个容器化应用，其中前端应用向API网关发起HTTP请求，API网关将接收到的请求数据通过Kafka转发给微服务。但我需要将处理结果作为响应返回给API网关，再回传给前端应用。我曾尝试使用普通的Producer和Consumer模式，这属于“发后即忘”的方式，无法回传请求响应。

原生产者代码：

using var p = new ProducerBuilder<string, string>(config).Build(); 
// Send the message to our test topic in Kafka
var dr = await p.ProduceAsync("test", message);

原消费者代码：

using var c = new ConsumerBuilder<Ignore, string>(conf).Build(); 
c.Subscribe("test"); 
// Consume a message from the test topic. 
var cr = c.Consume(cts.Token);

核心实现思路 #

要实现同步请求响应，核心是通过唯一请求ID+响应主题配对机制关联请求与响应：

• API网关发送请求时生成唯一请求ID，随消息传递给微服务，同时临时监听响应主题。
• 微服务处理完成后，携带相同请求ID将结果发回响应主题。
• API网关匹配请求ID获取响应，返回给前端。

具体实现步骤 #

1. 定义带请求ID的消息结构 #

请求和响应都需携带唯一请求ID，示例结构（可根据业务调整）：

// 请求消息
{
  "requestId": "uuid-123456",
  "data": "前端传递的请求内容"
}

// 响应消息
{
  "requestId": "uuid-123456",
  "result": "微服务处理后的结果",
  "status": "success/fail"
}

2. API网关改造（生产者+临时响应消费者） #

API网关在发送请求后，临时创建消费者监听响应主题，直到获取对应请求ID的响应：

// 生成唯一请求ID
var requestId = Guid.NewGuid().ToString();

// 构造请求消息
var requestPayload = JsonSerializer.Serialize(new 
{
    RequestId = requestId,
    Data = "前端请求数据" // 替换为实际接收的前端数据
});

var requestMessage = new Message<string, string>
{
    Key = requestId, // 用请求ID作为Key，方便后续过滤
    Value = requestPayload
};

// 发送请求到业务主题
using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
await producer.ProduceAsync("service-request-topic", requestMessage);

// 配置临时响应消费者
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "kafka:9092", // 替换为你的Kafka地址
    GroupId = $"gateway-temp-group-{requestId}", // 每个请求用独立消费组，避免干扰
    AutoOffsetReset = AutoOffsetReset.Latest // 只消费最新消息
};

using var responseConsumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
responseConsumer.Subscribe("service-response-topic");

// 设置超时时间，防止无限等待
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
try
{
    while (!cts.Token.IsCancellationRequested)
    {
        var consumeResult = responseConsumer.Consume(cts.Token);
        var response = JsonSerializer.Deserialize<dynamic>(consumeResult.Message.Value);
        
        // 匹配请求ID，获取对应响应
        if (response.RequestId == requestId)
        {
            return Results.Ok(new { response.Result, response.Status });
        }
    }
    // 超时未获取响应
    return Results.StatusCode(StatusCodes.Status504GatewayTimeout);
}
catch (OperationCanceledException)
{
    return Results.StatusCode(StatusCodes.Status504GatewayTimeout);
}
finally
{
    responseConsumer.Unsubscribe();
}

3. 微服务改造（消费者+响应生产者） #

微服务消费请求主题，处理后携带原请求ID发送响应到响应主题：

// 配置请求消费者
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    GroupId = "service-consumer-group",
    AutoOffsetReset = AutoOffsetReset.Earliest
};

using var requestConsumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
requestConsumer.Subscribe("service-request-topic");

// 配置响应生产者
using var responseProducer = new ProducerBuilder<string, string>(producerConfig).Build();

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

try
{
    while (!cts.Token.IsCancellationRequested)
    {
        var consumeResult = requestConsumer.Consume(cts.Token);
        var request = JsonSerializer.Deserialize<dynamic>(consumeResult.Message.Value);
        
        // 执行业务处理逻辑
        var processingResult = YourBusinessLogic(request.Data);
        
        // 构造响应消息
        var responsePayload = JsonSerializer.Serialize(new 
        {
            RequestId = request.RequestId,
            Result = processingResult,
            Status = "success"
        });
        
        var responseMessage = new Message<string, string>
        {
            Key = request.RequestId,
            Value = responsePayload
        };
        
        // 发送响应到响应主题
        await responseProducer.ProduceAsync("service-response-topic", responseMessage);
        
        // 提交消息偏移量
        requestConsumer.Commit(consumeResult);
    }
}
catch (OperationCanceledException)
{
    requestConsumer.Close();
}

4. 关键优化点 #

• 分区优化：将请求ID哈希后映射到固定分区，减少消费者需要遍历的消息数量。
• 资源回收：临时消费者必须在使用后取消订阅并释放，借助using语句确保资源不泄漏。
• 超时与重试：API网关设置合理超时时间，添加重试逻辑处理Kafka临时故障。
• 消息头存储请求ID：可将请求ID存入Kafka消息的Headers中，无需序列化到消息体，提升解析效率。

内容的提问来源于stack exchange，提问作者Malik Rashid Ahmad