聊聊如何利用kafka实现请求-响应模式

前言

在大多数场景中，我们经常使用kafka来做发布-订阅，在发布-订阅模型中，消息一旦发送就不再追踪后续处理，但在某些业务场景下，我们希望在发送消息后等待一个响应，然后根据这个响应来做我们后续的操作。在这种请求-响应模式，我们就可以利用spring kafka的ReplyingKafkaTemplate来实现

ReplyingKafkaTemplate

简介

ReplyingKafkaTemplate 是 Spring Kafka 中的一个高级特性，专门用于处理 Kafka 中的请求/响应模式。它允许你发送一个消息到 Kafka，并等待一个响应

使用场景

• 微服务间异步请求-响应： 当一个微服务需要从另一个微服务获取数据或执行操作，并希望在操作完成后得到通知时，可以使用 ReplyingKafkaTemplate。
• 状态查询： 如果一个服务需要定期或按需查询另一个服务的状态，但又不希望阻塞主线程等待响应，可以使用此模板。
• 异步任务确认： 当一个服务发起一个异步任务（如文件上传、计算任务等），并需要知道任务何时完成时，可以使用 ReplyingKafkaTemplate 来接收完成通知

如何使用

1、在项目中引入spring-kafka gav

 <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

2、 配置replyingKafkaTemplate bean

注： ReplyingKafkaTemplate需依赖ProducerFactory和KafkaMessageListenerContainer

配置示例如下

 /*** 创建一个repliesContainer* @param containerFactory* @return*/@Beanpublic ConcurrentMessageListenerContainer<String, String> repliesContainer(ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {// 和RecordHeader中的topic对应起来ConcurrentMessageListenerContainer<String, String> repliesContainer =containerFactory.createContainer(KafkaConstant.REPLY_TOPIC);repliesContainer.getContainerProperties().setGroupId("repliesGroup");repliesContainer.setAutoStartup(false);return repliesContainer;}/*** 创建一个replyingTemplate* @param pf* @param repliesContainer* @return*/@Beanpublic ReplyingKafkaTemplate<String, String, String> replyingTemplate(ProducerFactory<String, String> pf,ConcurrentMessageListenerContainer<String, String> repliesContainer) {ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate = new ReplyingKafkaTemplate<>(pf, repliesContainer);// 设置响应超时为10秒，默认5秒replyingKafkaTemplate.setReplyTimeout(10000);return replyingKafkaTemplate;}

3、producer发送请求并等待响应

  @SneakyThrows@Overridepublic String sendAndReceive(String topic, ParamRequest request)  {// 创建ProducerRecord类，用来发送消息ProducerRecord<String,String> producerRecord = new ProducerRecord<>(topic, JSONUtil.toJsonStr(request));// 添加KafkaHeaders.REPLY_TOPIC到record的headers参数中，这个参数配置我们想要转发到哪个Topic中producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, KafkaConstant.REPLY_TOPIC.getBytes()));// sendAndReceive方法返回一个Future类RequestReplyFuture，// 这里类里面包含了获取发送结果的Future类和获取返回结果的Future类。// 使用replyingKafkaTemplate发送及返回都是异步操作RequestReplyFuture<String, String, String> replyFuture = replyingKafkaTemplate.sendAndReceive(producerRecord);// 获取发送结果SendResult<String, String> sendResult = replyFuture.getSendFuture().get();log.info("send message success,topic:{},message:{},sendResult:{}",topic,JSONUtil.toJsonStr(request),sendResult.getRecordMetadata());// 获取响应结果ConsumerRecord<String, String> consumerRecord = replyFuture.get();String result = consumerRecord.value();log.info("result: {}",result);return result;}

注： 方法里都写了相应注释，就不再论述了

4、consumer进行监听，并将返回结果通过@SendTo转发回去

在官网贴了这么一段话

他的大意是为了支持@SendTo，侦听器容器工厂必须提供一个KafkaTemplate（在其replyTemplate属性中），用于发送回复。因此我们做如下配置

   @Bean@ConditionalOnMissingBeanpublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(KafkaTemplate kafkaTemplate, ConsumerFactory consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);factory.setReplyTemplate(kafkaTemplate);return factory;}

上述配置好后，就配置下监听器

 @KafkaListener(topics = TOPIC, groupId = "${lybgeek.consumer.group-id-prefix:lybgeek}-group-id", containerFactory = "kafkaListenerContainerFactory")/*** @SendTo 是一个Spring注解，常用于 Kafka 消费者方法之上，指示消息处理完成后应当将响应发送到哪个 Kafka 主题。* 使用场景：当你的应用作为服务端，需要对某个主题上的消息做出响应时，可以在处理该消息的方法上使用此注解来指定响应消息的目标主题。* 特点：简化了响应消息的路由配置，使得开发者无需显式地编写消息发送逻辑，只需关注业务处理逻辑。* 配合 ReplyingKafkaTemplate：在请求/响应模式中，@SendTo 指定的响应主题与 ReplyingKafkaTemplate 发送请求时设置的期望响应主题相匹配，从而使得请求方能够正确地接收响应消息。*///@SendTo("hello-test")@SendTopublic String listen(String data, Acknowledgment ack) {log.info("receive data:{}",data);if(JSONUtil.isJson(data)){Object result = execute(JSONUtil.toBean(data, ParamRequest.class),ack);if(result != null){ack.acknowledge();return JSONUtil.toJsonStr(result);};}return null;}

@SendTo的用途看我代码注释，具体用法可以查看官网
https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/annotation-send-to.html
进行了解

5、写个测试控制器

这个控制器的作用就是客户端发起http请求后，将请求参数送往kafka，kafka的消费方接收到http请求后，进行业务处理，并将业务结果通过kafka转发回去

  @PostMapping(value = "/**", consumes = {MediaType.APPLICATION_JSON_VALUE})
public Mono<ResponseEntity<byte[]>> forward(ProxyExchange<byte[]> proxy, Object params, HttpMethodEnum httpMethodEnum){try {String path = proxy.path().replace("/kafka", "").trim();ParamRequest paramRequest = buildParamRequest(path,params,httpMethodEnum);String topicCode = StringUtils.hasText(topicThreadLocal.get()) ? topicThreadLocal.get() : DEFAULT_TOPIC;String topic = TOPIC.replace(DEFAULT_TOPIC_PATTERN,topicCode);log.info(">>>>>>>>>>>>>> topic:{},httpMethod:{}, path:{},params:{}",topic,httpMethodEnum,path,params);Object result = kafkaService.sendAndReceive(topic,paramRequest);if(result != null){return Mono.just(ResponseEntity.ok(result.toString().getBytes()));}} catch (Exception e) {log.error(">>>>>>>>>>>>>> httpMethod:{},forward --> e:{}",httpMethodEnum.toString(),e.getMessage());} finally {topicThreadLocal.remove();}return Mono.just(ResponseEntity.ok(new byte[0]));}

核心就是这句代码

kafkaService.sendAndReceive(topic,paramRequest);

详细示例，可以查看文末的demo链接

使用ReplyingKafkaTemplate遇到的问题

No pending reply

这个问题是因为我拷贝了消费端配置文件，它配置了手动提交，而 ReplyingKafkaTemplate 是发送请求的一方，通常不需要特别的手动确认机制，因为 ReplyingKafkaTemplate 会等待响应或超时，因此改成自动确认即可

具体配置如下

spring:kafka:consumer:enable-auto-commit: ${KAFKA_CONSUMER_ENABLE_AUTO_COMMIT:true}

或者直接将消费端配置去掉也可以

总结

本文介绍通过ReplyingKafkaTemplate来实现请求-响应模式，在实际使用中，考虑到网络延迟和处理时间，调用ReplyingKafkaTemplate#sendAndReceive 方法可能会阻塞一段时间，因此在高负载环境下可能需要增加超时设置或使用回调机制

demo链接

https://github.com/lyb-geek/springboot-learning/tree/master/springboot-kafka-forward