如何用Kafka Streams实现请求-响应模式？Spring Kafka头丢失解决方案

我来帮你拆解这个问题：首先搞定Correlation ID丢失的问题，再一步步实现完整的请求-响应模式。

一、为什么Correlation ID会丢失？ #

Kafka Streams默认不会自动把原始消息的Headers传递到输出消息里——它的核心是处理Key-Value数据，Headers属于消息的元数据，需要你手动处理才能传递。

解决Correlation ID丢失的两种方案 #

方案1：用Processor/Transformer API手动传递Headers

这种方式适合在Streams内部完成消息转发，不需要额外依赖KafkaTemplate：

@Configuration
public class KafkaStreamsConfig {

    @Bean
    public KStream<String, Request> kStream(StreamsBuilder builder) {
        KStream<String, Request> requestStream = builder.stream("request-topic");

        requestStream.transform(() -> new Transformer<String, Request, KeyValue<String, Response>>() {
            private ProcessorContext context;

            @Override
            public void init(ProcessorContext context) {
                this.context = context;
            }

            @Override
            public KeyValue<String, Response> transform(String key, Request request) {
                // 从原始消息Headers中提取Correlation ID
                Header correlationIdHeader = context.headers().lastHeader("Correlation-ID");
                String correlationId = new String(correlationIdHeader.value(), StandardCharsets.UTF_8);

                // 处理请求生成响应
                Response response = processRequest(request);

                // 创建响应消息的Headers，把Correlation ID加进去
                Headers responseHeaders = new RecordHeaders();
                responseHeaders.add("Correlation-ID", correlationId.getBytes(StandardCharsets.UTF_8));

                // 转发消息到响应主题，带上自定义Headers
                context.forward(key, response, To.all().withHeaders(responseHeaders));

                return null; // 已经通过forward发送，无需返回值
            }

            @Override
            public void close() {}
        }).to("response-topic");

        return requestStream;
    }

    // 你的业务处理逻辑
    private Response processRequest(Request request) {
        return new Response(request.getId(), "处理完成：" + request.getContent());
    }
}

方案2：用foreachRecord结合KafkaTemplate发送响应

如果需要更灵活的发送控制（比如自定义分区、重试策略），可以用这种方式：

@Configuration
public class KafkaStreamsConfig {

    @Autowired
    private KafkaTemplate<String, Response> kafkaTemplate;

    @Bean
    public KStream<String, Request> kStream(StreamsBuilder builder) {
        KStream<String, Request> requestStream = builder.stream("request-topic");

        requestStream.foreachRecord(record -> {
            // 提取原始消息的Correlation ID
            String correlationId = new String(record.headers().lastHeader("Correlation-ID").value(), StandardCharsets.UTF_8);
            
            // 处理请求
            Response response = processRequest(record.value());

            // 构造响应消息，手动添加Correlation ID到Headers
            ProducerRecord<String, Response> responseRecord = new ProducerRecord<>("response-topic", record.key(), response);
            responseRecord.headers().add("Correlation-ID", correlationId.getBytes(StandardCharsets.UTF_8));
            
            // 发送响应
            kafkaTemplate.send(responseRecord);
        });

        return requestStream;
    }

    private Response processRequest(Request request) {
        return new Response(request.getId(), "处理完成：" + request.getContent());
    }
}

二、用Kafka Streams实现完整的请求-响应模式 #

请求-响应的核心是通过Correlation ID关联请求和响应：客户端发请求时带唯一ID，服务端处理后返回带相同ID的响应，客户端监听响应主题并匹配ID完成交互。

1. 客户端实现（Spring Boot） #

负责发送请求、监听响应、通过Correlation ID匹配结果：

@Service
public class RequestClient {
    @Autowired
    private KafkaTemplate<String, Request> kafkaTemplate;

    // 存储待处理的请求，key是Correlation ID
    private final ConcurrentHashMap<String, CompletableFuture<Response>> pendingRequests = new ConcurrentHashMap<>();

    public CompletableFuture<Response> sendRequest(Request request) {
        // 生成唯一的Correlation ID
        String correlationId = UUID.randomUUID().toString();
        CompletableFuture<Response> future = new CompletableFuture<>();
        pendingRequests.put(correlationId, future);

        // 构造请求消息，添加Correlation ID到Headers
        ProducerRecord<String, Request> record = new ProducerRecord<>("request-topic", request);
        record.headers().add("Correlation-ID", correlationId.getBytes(StandardCharsets.UTF_8));
        kafkaTemplate.send(record);

        // 设置超时，避免内存泄漏
        future.orTimeout(10, TimeUnit.SECONDS)
              .exceptionally(ex -> {
                  pendingRequests.remove(correlationId);
                  throw new RuntimeException("请求超时", ex);
              });
        return future;
    }

    // 监听响应主题，匹配Correlation ID返回结果
    @KafkaListener(topics = "response-topic")
    public void handleResponse(ConsumerRecord<String, Response> record) {
        String correlationId = new String(record.headers().lastHeader("Correlation-ID").value(), StandardCharsets.UTF_8);
        CompletableFuture<Response> future = pendingRequests.remove(correlationId);
        if (future != null) {
            future.complete(record.value());
        }
    }
}

2. 服务端实现（就是上面解决Correlation ID丢失的代码） #

服务端通过Kafka Streams消费请求、处理业务、带Correlation ID返回响应，和前面的方案一致。

注意事项 #

1. Correlation ID唯一性：一定要用UUID这类全局唯一的ID，避免不同请求的响应匹配错误；
2. 超时与内存清理：客户端必须设置超时，及时清理pendingRequests中的过期请求，防止内存泄漏；
3. 异常处理：服务端处理请求出错时，也要发送带Correlation ID的错误响应，让客户端知道请求失败；
4. 序列化配置：确保Request和Response的Serde配置正确，Spring Boot可以用JsonSerde自动完成JSON序列化。

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