前言
在大多数场景中，我们经常使用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
*/
@Bean
public 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
*/
@Bean
public 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
@Override
public 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转发回去
在官网贴了这么一段话
156247933a764f33a8156245d9d122e9_510604a1c8c352b88d22efd85dadff4f.png
他的大意是为了支持@SendTo，侦听器容器工厂必须提供一个KafkaTemplate（在其replyTemplate属性中），用于发送回复。因此我们做如下配置
@Bean
@ConditionalOnMissingBean
public 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")
@SendTo
public 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
