最近項(xiàng)目里有個(gè)需求，要消費(fèi)kafka里的數(shù)據(jù)。之前也手動(dòng)寫過代碼去消費(fèi)kafka數(shù)據(jù)。但是轉(zhuǎn)念一想。既然spring提供了消費(fèi)kafka的方法。就沒必要再去重復(fù)造輪子。于是嘗試使用spring的API。

項(xiàng)目技術(shù)背景，使用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文件里增加配置項(xiàng)，也可以單獨(dú)創(chuàng)建一個(gè)spring-context-XX.xml文件。

<!-- consumer configuration 該配置項(xiàng)可以根據(jù)自己業(yè)務(wù)的實(shí)際需求做增加或刪除--> <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> <!-- 自定義的消費(fèi)類，需要實(shí)現(xiàn)spring的接口 --> <bean /> <!-- 該類也是jar包里提供的，注入的監(jiān)聽類是自己定義的，topic名稱是配置文件引入的--> <bean class='org.springframework.kafka.listener.ContainerProperties'> <constructor-arg name='topics' value='${kafka.paypal.topic.name}'/> <property name='messageListener' ref='payPalConsumer' /> </bean> <!-- 改類也是jar里提供的，把這個(gè)containerProperties和consumerfactory 注入 --> <bean init-method='doStart'> <constructor-arg ref='consumerFactory' /> <constructor-arg ref='containerProperties' /> </bean>

2. 自定義消費(fèi)者類，消費(fèi)者類依然可以使用注解。

/** * 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); }}

使用這個(gè)步驟配置，一次性過。非常順利。

到此這篇關(guān)于spring 整合kafka監(jiān)聽消費(fèi)的配置過程的文章就介紹到這了,更多相關(guān)spring 整合kafka內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！