目的
软件需要使用什么技术都是按照业务逻辑来的嘛，那自动转发相对应的业务可以是什么呢？
可以使用转发功能实现业务解耦，系统A从Topic-A中获取到消息，进行处理后转发到Topic-B中，系统B监听Topic-B获取消息再次进行处理，
这个消息可以是订单相关数据，系统A处理用户提交的订单审核，系统B处理订单的物流信息等等。
实现方式
Spring-Kafka整合了两种消息转发方式：
使用Headers设置回复主题（Reply_Topic），这种方式比较特别，是一种请求响应模式，使用的是ReplyingKafkaTemplate类
手动转发，使用@SendTo注解将监听方法返回值转发到Topic中
@SendTo方式
SendTo的方式可以说是非常简单的，给我三秒钟，不不不，男人不可以这么快，三个钟好了，嘿嘿嘿。
配置ConcurrentKafkaListenerContainerFactory的ReplyTemplate
监听方法加上@SendTo注解
KafkaConfiguration.class
这里我们为监听容器工厂(ConcurrentKafkaListenerContainerFactory)配置一个ReplyTemplate，ReplyTemplate是我们用来转发消息所使用的类。
@SendTo注解本质其实就是利用这个ReplyTemplate转发监听方法的返回值到对应的Topic中，
我们也可以是用代码实现KakfaTemplate.send()，不过使用注解的好处就是减少代码量，加快开发效率。
@Bean public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setReplyTemplate(kafkaTemplate()); return factory; }
@Component public class ForwardListener { private static final Logger log= LoggerFactory.getLogger(ForwardListener.class); @KafkaListener(id = "forward", topics = "topic.quick.target") @SendTo("topic.quick.real") public String forward(String data) { log.info("topic.quick.target forward "+data+" to topic.quick.real"); return "topic.quick.target send msg : " + data; } }
顺便就写个测试方法测试一下吧，可以看到运行成功后，
topic.quick.real这个主题会产生一条数据，这条数据就是我们在forward方法返回的值。
@Autowired private KafkaTemplate kafkaTemplate; @Test public void testForward() { kafkaTemplate.send("topic.quick.target", "test @SendTo"); }
ReplyTemplate方式
使用ReplyTemplate方式不同于@SendTo方式，@SendTo是直接将监听方法的返回值转发对应的Topic中，
而ReplyTemplate也是将监听方法的返回值转发Topic中，但转发Topic成功后，会被请求者消费。
这是怎么回事呢？我们可以回想一下请求响应模式，这种模式其实我们是经常使用的，
就像你调用某个第三方接口，这个接口会把响应报文返回给你，你可以根据业务处理这段响应报文。
而ReplyTemplate方式的这种请求响应模式也是相同的，
首先生成者发送消息到Topic-A中，Topic-A的监听器则会处理这条消息，紧接着将消息转发到Topic-B中，
当这条消息转发到Topic-B成功后则会被ReplyTemplate接收。那最终消费者获得的是被处理过的数据。
ReplyTemplate实现的代码也并不复杂，实现的功能确更多。
讲一下流程吧：
配置ConcurrentKafkaListenerContainerFactory的ReplyTemplate
配置topic.quick.request的监听器
注册一个KafkaMessageListenerContainer类型的监听容器，监听topic.quick.reply，这个监听器里面我们不处理任何事情，交由ReplyingKafkaTemplate处理
通过ProducerFactory和KafkaMessageListenerContainer创建一个ReplyingKafkaTemplate类型的Bean，设置回复超时时间为10秒
@Bean public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setReplyTemplate(kafkaTemplate()); return factory; }
@KafkaListener(id = "replyConsumer", topics = "topic.quick.request",containerFactory = "kafkaListenerContainerFactory") @SendTo public String replyListen(String msgData){ log.info("topic.quick.request receive : "+msgData); return "topic.quick.reply reply : "+msgData; } @Bean public KafkaMessageListenerContainer<String, String> replyContainer(@Autowired ConsumerFactory consumerFactory) { ContainerProperties containerProperties = new ContainerProperties("topic.quick.reply"); return new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); } @Bean public ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate(@Autowired ProducerFactory producerFactory, KafkaMessageListenerContainer replyContainer) { ReplyingKafkaTemplate template = new ReplyingKafkaTemplate<>(producerFactory, replyContainer); template.setReplyTimeout(10000); return template; }
发送消息就显得稍微有点麻烦了，不过在项目编码过程中可以把它封装成一个工具类调用。
我们需要创建ProducerRecord类，用来发送消息，并添加KafkaHeaders.REPLY_TOPIC到record的headers参数中，这个参数配置我们想要转发到哪个Topic中。
使用replyingKafkaTemplate.sendAndReceive()方法发送消息，该方法返回一个Future类RequestReplyFuture，
这里类里面包含了获取发送结果的Future类和获取返回结果的Future类。使用replyingKafkaTemplate发送及返回都是异步操作。
调用RequestReplyFuture.getSendFutrue().get()方法可以获取到发送结果
调用RequestReplyFuture.get()方法可以获取到响应结果
@Autowired private ReplyingKafkaTemplate replyingKafkaTemplate; @Test public void testReplyingKafkaTemplate() throws ExecutionException, InterruptedException, TimeoutException { ProducerRecord<String, String> record = new ProducerRecord<>("topic.quick.request", "this is a message"); record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, "topic.quick.reply".getBytes())); RequestReplyFuture<String, String, String> replyFuture = replyingKafkaTemplate.sendAndReceive(record); SendResult<String, String> sendResult = replyFuture.getSendFuture().get(); System.out.println("Sent ok: " + sendResult.getRecordMetadata()); ConsumerRecord<String, String> consumerRecord = replyFuture.get(); System.out.println("Return value: " + consumerRecord.value()); Thread.sleep(20000); }
注意：
由于ReplyingKafkaTemplate也是通过监听容器实现的，所以响应时间可能会较慢，要注意选择合适的场景使用。