内容简介：Spring Apache Kafka (spring-kafka)提供了基于卡夫卡的消息传递解决方案的高级抽象。传统的请求响应模型中，响应容易被堵塞，造成两个系统耦合，调用者需要等待到响应返回才能继续做自己的工作，这在分布式系统中，流量比较大情况下几乎不现实，使用消息模型只能每次请求一个消息，响应再来一个消息，用两个消息组合成请求响应，虽然编程没有传统请求响应方便，但是系统松耦合，相互协调好。spring-kafka使用起来了很简单:引入Maven包：
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即可有用。
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网
猜你喜欢:
- SpringSession：请求与响应重写
- 谈谈HTTP的请求和响应
- django从请求到响应的过程
- 关于HTTP报文请求方法和状态响应码
- 优雅的读取http请求或响应的数据
- 优雅的记录http请求或响应的数据
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
遗传算法与工程优化
程润伟 / 清华大学出版社 / 2004-1 / 39.00元
《遗传算法与工程优化》总结了遗传算法在工业工程相关领域应用的前沿进展。全书共分9章：遗传算法基础、组合优化问题、多目标优化问题、模糊优化问题、可靠性设计问题、调度问题、高级运输问题、网络设计与路径问题和制造元设计问题。内容既涵盖了遗传算法在传统优化问题中的新进展，又涉及了目前在供应链和物流研究中相当热门的话题。一起来看看 《遗传算法与工程优化》 这本书的介绍吧!
