转载

使用Spring Boot的Kafka消息机制实现请求响应模型

Spring Apache Kafka (spring-kafka)提供了基于卡夫卡的消息传递解决方案的高级抽象。传统的请求响应模型中，响应容易被堵塞，造成两个系统耦合，调用者需要等待到响应返回才能继续做自己的工作，这在分布式系统中，流量比较大情况下几乎不现实，使用消息模型只能每次请求一个消息，响应再来一个消息，用两个消息组合成请求响应，虽然编程没有传统请求响应方便，但是系统松耦合，相互协调好。

spring-kafka使用起来了很简单:

引入Maven包：

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

消息生产者代码：

@Autowired
KafkaTemplate<String, String> kafkaTemplate;

 <b>public</b> <b>void</b> send(@RequestParam String productId) {
        kafkaTemplate.send(<font>"cdProduct"</font><font>, productId);
  }
</font>

生产者的配置application.property：

spring.kafka.consumer.group-id=ecomm
spring.kafka.bootstrap-servers=localhost:9092

消息接受者代码，这里接受到productId以后，查询到Product对象，再给发送者发回去，模拟请求-响应模型：

@KafkaListener(topics = <font>"cdProduct"</font><font>)
    <b>public</b> <b>void</b> onAction(ConsumerRecord<?, ?> consumerRecord) {
        System.out.printf(</font><font>"接受到="</font><font> + consumerRecord);
        String productId = (String) consumerRecord.value();
        System.out.printf(</font><font>"接受到productId="</font><font> + productId);
        Product product = productRepo.findById(productId).orElse(<b>new</b> Product());
        kafkaTemplate.send(</font><font>"cdProductReply"</font><font>, product);
    }
</font>

消费者需要配置JSON序列化将Product变成JSON，这里只要配置在application.property中即可，无需做代码生成自己ProducerFactory工厂：

spring.kafka.consumer.group-id=ecomm
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

生产者接受到消费者查询到的Product放入自己的缓存。

@KafkaListener(topics = <font>"cdProductReply"</font><font>)
    <b>public</b> <b>void</b> onAction(Product product) {
        System.out.printf(</font><font>"接受到新的Product"</font><font> + product.getName());
        cache.put(product.getId(), product);

    }
</font>

为了实现Product直接序列化接受，需要在自己的入口类Application中加入：

@Bean
<b>public</b> StringJsonMessageConverter jsonConverter() {
        <b>return</b> <b>new</b> StringJsonMessageConverter();
}

无需配置监听器连接工厂ConcurrentKafkaListenerContainerFactory即可有用。

原文 https://www.jdon.com/51213

正文到此结束

所属分类： Spring

本文标签： 分布式 CTO message id maven 代码监听器 IO db Bootstrap http cache tk apache 配置 consumer bean 缓存 spring cat list 做自己 find value Action 模型 js json Property 分布式系统 https App producer
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

使用Spring Boot的Kafka消息机制实现请求响应模型

热门推荐

相关文章

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流