spring整合kafka（配置文件方式 消費者）

Kafka官方文檔有   https://docs.spring.io/spring-kafka/reference/htmlsingle/

這里是配置文件實現的方式

先引入依賴

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


創建 spring-context-kafka-consumer.xml     當然要配置spring掃描該配置文件

配置文件里邊內容如下

<!-- 基本的配置參數 可以寫成配置文件或者這種${bootstrap.servers} 配置文件獲取的 可以區分開發測試環境 -->
<bean id="consumerProperties" class="java.util.HashMap">
 <constructor-arg>
 <map>
 <entry key="bootstrap.servers" value="${bootstrap.servers}" />
 <entry key="group.id" value="0" />
 <entry key="enable.auto.commit" value="true" />
 <entry key="auto.commit.interval.ms" value="1000" />
 <entry key="session.timeout.ms" value="15000" />
 <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>

<!-- 創建工廠 然后把配置信息注入-->
<bean id="consumerFactory"
 class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
 <constructor-arg>
 <ref bean="consumerProperties" />
 </constructor-arg>
</bean>

<!-- 把實際消費的類關聯進來 -->
<bean id="messageListernerConsumerService" class="com.test.kafkaConsumer.KafkaConsumer" />

<!-- 然后把這個類和消費的topic注入這個container topic也配置成靈活的 -->
<bean id="containerProperties"
 class="org.springframework.kafka.listener.config.ContainerProperties">
 <constructor-arg name="topics" value="${topic}"/>
 <property name="messageListener" ref="messageListernerConsumerService" />
</bean>

<!-- 把這個container和factory 注入 -->
<bean id="messageListenerContainer"
 class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
 init-method="doStart">
 <constructor-arg ref="consumerFactory" />
 <constructor-arg ref="containerProperties" />
</bean>

<!-- 這個可以配置一個類消費多個topic   如果需要不同的類消費不同的topic 就配置多個container關聯不通的類 -->


消費的類

public class KafkaConsumer implements MessageListener<Integer, String> {

 

 @Override
 public void onMessage(ConsumerRecord<Integer, String> record) {
 
 String value = record.value();
　　}

//因為配置文件已經關聯這個類 所以只要啟動spring項目 就可以監聽消費配置的topic value就是推送過來的消息

}

 
更多實時計算,Kafka等相關技術博文，歡迎關注實時流式計算