ReplyingKafkaTemplate动态配置request/reply主题实现咨询

• 实现Kafka同步请求-响应业务：生产者向requesttopic发送消息后，等待消费者处理消息，再将响应发送至requestreplytopic返回。
• 期望根据前端用户发起的请求动态生成请求主题与对应响应主题：例如用户参与商品出价时，使用对应商品ID作为请求主题，${productid}_resp作为对应的响应主题。该场景下会有多条出价消息发送至消息Broker，响应需返回至前端UI。
• 计划以商品ID作为主题名，后续需要查询某商品的全部出价记录时，从头读取对应主题的所有消息。

• 当前业务场景是否适配ReplyingKafkaTemplate
• ReplyingKafkaTemplate是否支持动态创建配置

回复容器配置 #

@Value("${kafka.topic.requestreply-topic}")
private String requestReplyTopic;

@Bean
public <T> KafkaMessageListenerContainer<String, ResponseBase<T>> replyContainer(
        ConsumerFactory<String, ResponseBase<T>> cf) {
    ContainerProperties containerProperties = new ContainerProperties(requestReplyTopic);
    return new KafkaMessageListenerContainer<>(cf, containerProperties);
}

消费者监听逻辑 #

@KafkaListener(topics = "${kafka.topic.request-topic}", errorHandler = "listen3ErrorHandler")
@SendTo
public ResponseBase<Object> listen(CreateBidRequest request) throws BidException {
    Product product = productRepository.findById(request.getProductId()).orElseThrow(() -> new BidException(ErrorCode.INVALID_PRODUCT_ID));
    Bid bid = Bid.builder().bidId(UUID.randomUUID().toString()).bidAmount(request.getBidAmount()).product(product).build();
    bidRepository.save(bid);
    return ResponseBase.builder().response(bid).build();
}

• 你的业务场景不适配默认固定配置的ReplyingKafkaTemplate。原生ReplyingKafkaTemplate初始化时会绑定固定的回复监听容器，容器仅能监听预先配置的固定主题，无法直接匹配动态生成${productid}_resp这类多回复主题的需求。另外你按商品ID拆分请求主题、需要回溯单主题全量消息的设计，和ReplyingKafkaTemplate默认单请求/回复主题、靠correlationId匹配请求响应的底层逻辑不匹配，硬改适配成本极高。
• ReplyingKafkaTemplate本身支持动态创建，但完全不推荐在该场景下使用：如果要适配动态主题，每新增一个商品对应的主题，就需要手动创建专属的回复监听容器、初始化新的ReplyingKafkaTemplate实例，一旦商品量级上涨，会生成大量监听容器，占用过多客户端和Broker连接资源，资源开销和运维成本都会失控。
• 针对该场景更合理的实现方式：
  • 不要按商品ID拆分Kafka主题。Kafka集群对主题总数有阈值限制，单集群主题数过万就会明显影响性能，按商品ID拆分主题的方案长期运行必然遇到性能瓶颈。查询某商品全部出价记录的需求，直接查询已经持久化的Bid数据库表即可，不需要从头读取Kafka主题——Kafka是消息中间件不是业务数据存储载体，不适合作为持久化查询的数据源。
  • 使用统一的公共请求主题、公共回复主题即可满足需求。生产者发消息时在消息头传入唯一correlationId、指定回复目标为公共回复主题，消费者处理完成后把correlationId原封不动带回响应，ReplyingKafkaTemplate会自动匹配对应请求返回结果给前端，完全能支撑出价请求的同步响应需求。
  • 如果一定要保留动态回复主题的逻辑，不需要硬改ReplyingKafkaTemplate配置，发消息时直接在消息头设置KafkaHeaders.REPLY_TOPIC为动态生成的主题名即可，代码中使用的@SendTo注解会自动识别该消息头，把响应发到指定主题，不需要提前在配置里硬编码主题。

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