美文网首页Apache KafkaSpring-Kafka史上最强入门教程消息中间件
Spring-Kafka(七)—— 实现消息转发以及ReplyT

Spring-Kafka(七)—— 实现消息转发以及ReplyT

作者: 海苔胖胖 | 来源:发表于2018-09-12 14:53 被阅读200次

    目的

    软件需要使用什么技术都是按照业务逻辑来的嘛,那自动转发相对应的业务可以是什么呢?

    可以使用转发功能实现业务解耦,系统A从Topic-A中获取到消息,进行处理后转发到Topic-B中,系统B监听Topic-B获取消息再次进行处理,这个消息可以是订单相关数据,系统A处理用户提交的订单审核,系统B处理订单的物流信息等等。

    实现方式

    Spring-Kafka整合了两种消息转发方式:

    1. 使用Headers设置回复主题(Reply_Topic),这种方式比较特别,是一种请求响应模式,使用的是ReplyingKafkaTemplate类
    2. 手动转发,使用@SendTo注解将监听方法返回值转发到Topic中



    @SendTo方式

    SendTo的方式可以说是非常简单的,给我三秒钟,不不不,男人不可以这么快,三个钟好了,嘿嘿嘿。

    1. 配置ConcurrentKafkaListenerContainerFactory的ReplyTemplate
    2. 监听方法加上@SendTo注解
    3. 还想有第三步?回家睡觉吧
    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实现的代码也并不复杂,实现的功能确更多。

    讲一下流程吧:

    1. 配置ConcurrentKafkaListenerContainerFactory的ReplyTemplate
    2. 配置topic.quick.request的监听器
    3. 注册一个KafkaMessageListenerContainer类型的监听容器,监听topic.quick.reply,这个监听器里面我们不处理任何事情,交由ReplyingKafkaTemplate处理
    4. 通过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;
        }
    

    发送消息就显得稍微有点麻烦了,不过在项目编码过程中可以把它封装成一个工具类调用。

    1. 我们需要创建ProducerRecord类,用来发送消息,并添加KafkaHeaders.REPLY_TOPIC到record的headers参数中,这个参数配置我们想要转发到哪个Topic中。
    2. 使用replyingKafkaTemplate.sendAndReceive()方法发送消息,该方法返回一个Future类RequestReplyFuture,这里类里面包含了获取发送结果的Future类和获取返回结果的Future类。使用replyingKafkaTemplate发送及返回都是异步操作。
    3. 调用RequestReplyFuture.getSendFutrue().get()方法可以获取到发送结果
    4. 调用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也是通过监听容器实现的,所以响应时间可能会较慢,要注意选择合适的场景使用。

    相关文章

      网友评论

        本文标题:Spring-Kafka(七)—— 实现消息转发以及ReplyT

        本文链接:https://www.haomeiwen.com/subject/hbbmgftx.html