av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術文章
文章詳情頁

spring 整合kafka監聽消費的配置過程

瀏覽:25日期:2023-07-20 15:01:58
前言

最近項目里有個需求,要消費kafka里的數據。之前也手動寫過代碼去消費kafka數據。但是轉念一想。既然spring提供了消費kafka的方法。就沒必要再去重復造輪子。于是嘗試使用spring的API。

項目技術背景,使用springMVC,XML配置和注解相互使用。kafka的配置都是使用XML方式。

整合過程

1. 引入spring-kafka的依賴包

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

2. 在spring的xml文件里增加配置項,也可以單獨創建一個spring-context-XX.xml文件。

<!-- consumer configuration 該配置項可以根據自己業務的實際需求做增加或刪除--> <bean class='java.util.HashMap'> <constructor-arg> <map><entry key='bootstrap.servers' value='${kafka.bootstrap.servers}' /><entry key='group.id' value='group' /><entry key='enable.auto.commit' value='true' /><entry key='auto.commit.interval.ms' value='3000' /><entry key='session.timeout.ms' value='10000' /><entry key='key.deserializer' value='org.apache.kafka.common.serialization.StringDeserializer' /><entry key='value.deserializer' value='org.apache.kafka.common.serialization.StringDeserializer' /> </map> </constructor-arg> </bean> <!-- create factory 該類是spring jar包里提供,就這么配置--> <bean class='org.springframework.kafka.core.DefaultKafkaConsumerFactory'> <constructor-arg> <ref bean='consumerProperties' /> </constructor-arg> </bean> <!-- 自定義的消費類,需要實現spring的接口 --> <bean /> <!-- 該類也是jar包里提供的,注入的監聽類是自己定義的,topic名稱是配置文件引入的--> <bean class='org.springframework.kafka.listener.ContainerProperties'> <constructor-arg name='topics' value='${kafka.paypal.topic.name}'/> <property name='messageListener' ref='payPalConsumer' /> </bean> <!-- 改類也是jar里提供的,把這個containerProperties和consumerfactory 注入 --> <bean init-method='doStart'> <constructor-arg ref='consumerFactory' /> <constructor-arg ref='containerProperties' /> </bean>

2. 自定義消費者類,消費者類依然可以使用注解。

/** * get msg from kafka */@Component public class PayPalConsumer implements MessageListener<String, String> { private static Logger logger = LoggerFactory.getLogger(PayPalConsumer.class); @Autowired private XXService XXService; @Override public void onMessage(ConsumerRecord<String, String> authorizeRecord) { String value = authorizeRecord.value(); if (StringUtils.isEmpty(value)){ logger.warn('receive message from kafka is null'); return; } logger.info('receive message from kafka is {}',value); }}

使用這個步驟配置,一次性過。非常順利。

到此這篇關于spring 整合kafka監聽消費的配置過程的文章就介紹到這了,更多相關spring 整合kafka內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網!

標簽: Spring
相關文章:
主站蜘蛛池模板: 久久久久91| 国产视频一区二区 | 日韩成人免费视频 | 男人的天堂一级片 | 精品一区二区三区在线视频 | 亚洲精品视频在线看 | 97伦理影院 | 99精品久久久久久 | 中文字幕视频在线观看 | 国产日韩欧美在线观看 | 91视频a| 免费观看一级毛片视频 | 日韩在线视频网址 | 亚洲第一成年免费网站 | 一级黄色影片在线观看 | 久久久99精品免费观看 | 亚洲精品国产成人 | 亚洲免费视频播放 | 国产在线精品一区二区 | 欧美一区二区三区在线播放 | 一区二区成人 | 婷婷精品| 亚洲精品不卡 | 久久久精品一区二区三区四季av | 国产成人精品一区二区 | 久久久久亚洲精品 | 国产黄色精品 | 一区在线视频 | 国产亚洲一区精品 | 成人国产精品入口免费视频 | 欧美一区二区激情三区 | 国产精品96久久久久久 | 日韩一区二区在线观看视频 | 中日韩欧美一级片 | 国产精品国产精品国产专区不片 | 日一区二区三区 | 久草网址 | 久草在线中文888 | 久久综合色综合 | 国产精品视频在 | 337p日本欧洲亚洲大胆精蜜臀 |