最近項目里有個需求，要消費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內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！