系列目錄
kafka原理和實踐（三）spring-kafka生產者源碼
kafka原理和實踐（四）spring-kafka消費者源碼
=========正文分割線====================
作為一個MQ做基本的功能自然就是消息的生產和消費，本章以XML配置的方式實現消息的生產和消費。
一、生產者
1.1.配置
spring-kafka 提供了org.springframework.kafka.core.KafkaTemplate
xml配置如下，producerProperties中的具體配置暫時不用在意，后面有一章專門講xml配置：
1 <?xml version="1.0" encoding="UTF-8"?> 2 <beans xmlns="http://www.springframework.org/schema/beans" 3 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" 4 xsi:schemaLocation="http://www.springframework.org/schema/beans 5 http://www.springframework.org/schema/beans/spring-beans-3.0.xsd 6 http://www.springframework.org/schema/context 7 http://www.springframework.org/schema/context/spring-context.xsd"> 8 <context:property-placeholder location="classpath*:config/application.properties" /> 9 <!-- 定義producer的參數 --> 10 <bean id="producerProperties" class="java.util.HashMap"> 11 <constructor-arg> 12 <map> 13 <entry key="bootstrap.servers" value="${bootstrap.servers}" /> 14 <entry key="group.id" value="${group.id}" /> 15 <entry key="retries" value="${retries}" /> 16 <entry key="batch.size" value="${batch.size}" /> 17 <entry key="linger.ms" value="${linger.ms}" /> 18 <entry key="buffer.memory" value="${buffer.memory}" /> 20 <entry key="acks" value="${acks}" /> 22 <entry key="key.serializer" 23 value="org.apache.kafka.common.serialization.StringSerializer" /> 24 <entry key="value.serializer" 25 value="org.apache.kafka.common.serialization.StringSerializer" /> 26 </map> 27 </constructor-arg> 28 </bean> 29 30 <!-- 創建kafkatemplate需要使用的producerfactory bean --> 31 <bean id="producerFactory" 32 class="org.springframework.kafka.core.DefaultKafkaProducerFactory"> 33 <constructor-arg> 34 <ref bean="producerProperties" /> 35 </constructor-arg> 36 </bean> 37 38 <!-- 創建kafkatemplate bean，使用的時候，只需要注入這個bean，即可使用template的send消息方法 --> 39 <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate"> 40 <constructor-arg ref="producerFactory" /> 41 <constructor-arg name="autoFlush" value="true" /> 42 <property name="defaultTopic" value="default" /> 43 </bean> 46 </beans>
如上圖，xml主要配置了KafkaTemplate的構造參數producerFactory和autoFlush，對應了一個KafkaTemplate源碼中的2參構造函數。
producerProperties：設置生產者公產需要的配置
producerFactory：定義了生產者工廠構造方法
kafkaTemplate：定義了使用producerFactory和是否自動刷新，2個參數來構造kafka生產者模板類。
1.2 發送kafka消息
1.根據topic、partition、key發送數據data。
2.接收ListenableFuture添加成功、失敗回調函數。
1 ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("topic", "partition","key","data"); 2 //發送成功回調 3 SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() { 4 @Override 5 public void onSuccess(SendResult<String, String> result) { 6 //成功業務邏輯 7 } 8 } 9 //發送失敗回調 10 FailureCallback failureCallback = new FailureCallback() { 11 @Override 12 public void onFailure(Throwable ex) { 13 //失敗業務邏輯 14 } 15 } 16 listenableFuture.addCallback(successCallback, failureCallback);
二、消費者
2.1配置
consumerProperties中的具體配置暫時不用在意，后面有一章專門講xml配置
1 <!-- 1.定義consumer的參數 --> 2 <bean id="consumerProperties" class="java.util.HashMap"> 3 <constructor-arg> 4 <map> 5 <entry key="bootstrap.servers" value="${bootstrap.servers}" /> 6 <entry key="group.id" value="${group.id}" /> 7 <entry key="enable.auto.commit" value="${enable.auto.commit}" /> 8 <entry key="session.timeout.ms" value="${session.timeout.ms}" /> 9 <entry key="key.deserializer" 10 value="org.apache.kafka.common.serialization.StringDeserializer" /> 11 <entry key="value.deserializer" 12 value="org.apache.kafka.common.serialization.StringDeserializer" /> 13 </map> 14 </constructor-arg> 15 </bean> 16 17 <!-- 2.創建consumerFactory bean --> 18 <bean id="consumerFactory" 19 class="org.springframework.kafka.core.DefaultKafkaConsumerFactory" > 20 <constructor-arg> 21 <ref bean="consumerProperties" /> 22 </constructor-arg> 23 </bean> 24 25 <!-- 3.定義消費實現類 --> 26 <bean id="kafkaConsumerService" class="xxx.service.impl.KafkaConsumerSerivceImpl" /> 27 28 <!-- 4.消費者容器配置信息 --> 29 <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties"> 30 <!-- topic --> 31 <constructor-arg name="topics"> 32 <list> 33 <value>${kafka.consumer.topic.credit.for.lease}</value> 34 <value>${loan.application.feedback.topic}</value> 35 <value>${templar.agreement.feedback.topic}</value> 36 <value>${templar.aggrement.active.feedback.topic}</value> 37 <value>${templar.aggrement.agreementRepaid.topic}</value> 38 <value>${templar.aggrement.agreementWithhold.topic}</value> 39 <value>${templar.aggrement.agreementRepayRemind.topic}</value> 40 </list> 41 </constructor-arg> 42 <property name="messageListener" ref="kafkaConsumerService" /> 43 </bean> 44 <!-- 5.消費者並發消息監聽容器，執行doStart()方法 --> 45 <bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart" > 46 <constructor-arg ref="consumerFactory" /> 47 <constructor-arg ref="containerProperties" /> 48 <property name="concurrency" value="${concurrency}" /> 49 </bean>
1.consumerProperties-》consumerFactory 載入配置構造消費者工廠
2.messageListener-》containerProperties 載入容器配置（topics）
3.consumerFactory+containerProperties-》messageListenerContainer 容器配置（topics）+消息監聽器，構造一個並發消息監聽容器，並執行初始化方法doStart
2.2消費kafka消息
方案1:直接實現MessageListener接口，復寫onMessage方法，實現自定義消費業務邏輯。
1 public class KafkaConsumerSerivceImpl implements MessageListener<String, String> { 2 @Override 3 public void onMessage(ConsumerRecord<String, String> data) { 4 //根據不同主題，消費 5 if("主題1".equals(data.topic())){ 6 //邏輯1 7 }else if("主題2".equals(data.topic())){ 8 //邏輯2 9 } 10 } 11 }
方案2：使用@KafkaListener注解，並設置topic,支持SPEL表達式。這樣方便拆分多個不同topic處理不同業務邏輯。（特別是有自己的事務的時候，尤其方便）
1 import org.springframework.kafka.annotation.KafkaListener; 2 3 public class KafkaConsumerSerivceImpl { 4 @KafkaListener(topics = "${templar.aggrement.agreementWithhold.topic}") 5 void templarAgreementNoticewithhold(ConsumerRecord<String, String> data){ 6 //消費業務邏輯 7 } 8 }
三、總結
本章我們實現了一個簡單的kafka生產、消費消息的實踐。到這里我們已經會基本使用kafka了。是不是很簡單...
下一章，我們從源碼角度來深入分析spring-kafka。