久久r热视频,国产午夜精品一区二区三区视频,亚洲精品自拍偷拍,欧美日韩精品二区

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

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

瀏覽:18日期: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
相關文章:
主站蜘蛛池模板: 牡丹江市| 武清区| 曲阜市| 深泽县| 福建省| 黄梅县| 三台县| 凤翔县| 昭通市| 鸡西市| 运城市| 柏乡县| 修水县| 东兰县| 扎鲁特旗| 石门县| 平谷区| 枣阳市| 玛多县| 宁武县| 枣庄市| 县级市| 文山县| 醴陵市| 苍山县| 独山县| 晋城| 光泽县| 石楼县| 康马县| 东乌| 康保县| 旬邑县| 昌乐县| 岢岚县| 册亨县| 上饶市| 苏尼特右旗| 桐庐县| 荃湾区| 灵宝市|