kafka原理和实践（二）spring-kafka简单实践

系列目录

kafka原理和实践（一）原理：10分钟入门

kafka原理和实践（二）spring-kafka简单实践

kafka原理和实践（三）spring-kafka生产者源码

kafka原理和实践（四）spring-kafka消费者源码

kafka原理和实践（五）spring-kafka配置详解

kafka原理和实践（六）总结升华

=========正文分割线====================

作为一个MQ做基本的功能自然就是消息的生产和消费，本章以XML配置的方式实现消息的生产和消费。

一、生产者

1.1.配置

spring-kafka 提供了org.
springframework.
kafka.
core.KafkaTemplate

xml配置如下，
producerProperties中的具体配置暂时不用在意，后面有一章专门讲xml配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
11 <?xml version="1.0" encoding="UTF-8"?>

2 2 <beans xmlns="http://www.springframework.org/schema/beans"

3 3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"

4 4     xsi:schemaLocation="http://www.springframework.org/schema/beans  

5 5          http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  

6 6          http://www.springframework.org/schema/context  

7 7          http://www.springframework.org/schema/context/spring-context.xsd">

8 8     <context:property-placeholder location="classpath*:config/application.properties" />

9 9     <!-- 定义producer的参数 -->

1010     <bean id="producerProperties" class="java.util.HashMap">

1111         <constructor-arg>

1212             <map>

1313                 <entry key="bootstrap.servers" value="${bootstrap.servers}" />

1414                 <entry key="group.id" value="${group.id}" />

1515                 <entry key="retries" value="${retries}" />

1616                 <entry key="batch.size" value="${batch.size}" />

1717                 <entry key="linger.ms" value="${linger.ms}" />

1818                 <entry key="buffer.memory" value="${buffer.memory}" />            

1920                 <entry key="acks" value="${acks}" />              

2022                 <entry key="key.serializer"

2123                     value="org.apache.kafka.common.serialization.StringSerializer" />

2224                 <entry key="value.serializer"

2325                     value="org.apache.kafka.common.serialization.StringSerializer" />

2426             </map>

2527         </constructor-arg>

2628     </bean>

2729 

2830     <!-- 创建kafkatemplate需要使用的producerfactory bean -->

2931     <bean id="producerFactory"

3032         class="org.springframework.kafka.core.DefaultKafkaProducerFactory">

3133         <constructor-arg>

3234             <ref bean="producerProperties" />

3335         </constructor-arg>

3436     </bean>

3537 

3638     <!-- 创建kafkatemplate bean，使用的时候，只需要注入这个bean，即可使用template的send消息方法 -->

3739     <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">

3840         <constructor-arg ref="producerFactory" />

3941         <constructor-arg name="autoFlush" value="true" />

4042         <property name="defaultTopic" value="default" />

4143     </bean>

4246 </beans>  

43

如上图，xml主要配置了KafkaTemplate的构造参数producerFactory和autoFlush，对应了一个KafkaTemplate源码中的2参构造函数。

producerProperties：设置生产者公产需要的配置
producerFactory：定义了生产者工厂构造方法
kafkaTemplate：定义了使用producerFactory和是否自动刷新，2个参数来构造kafka生产者模板类。

1.2  发送kafka消息

1.根据topic、partition、key发送数据data。

2.接收ListenableFuture添加成功、失败回调函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
11 ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("topic", "partition","key","data");

2 2 //发送成功回调

3 3 SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {

4 4     @Override

5 5     public void onSuccess(SendResult<String, String> result) {

6 6        //成功业务逻辑

7 7     }

8 8 }

9 9 //发送失败回调

1010 FailureCallback failureCallback = new FailureCallback() {

1111     @Override

1212     public void onFailure(Throwable ex) {

1313         //失败业务逻辑

1414     }

1515 }

1616 listenableFuture.addCallback(successCallback, failureCallback);

17

二、消费者

2.1配置

1
2
1consumerProperties中的具体配置暂时不用在意，后面有一章专门讲xml配置

2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
11 <!-- 1.定义consumer的参数 -->

2 2     <bean id="consumerProperties" class="java.util.HashMap">

3 3         <constructor-arg>

4 4             <map>

5 5                 <entry key="bootstrap.servers" value="${bootstrap.servers}" />

6 6                 <entry key="group.id" value="${group.id}" />

7 7                 <entry key="enable.auto.commit" value="${enable.auto.commit}" />

8 8                 <entry key="session.timeout.ms" value="${session.timeout.ms}" />

9 9                 <entry key="key.deserializer"

1010                     value="org.apache.kafka.common.serialization.StringDeserializer" />

1111                 <entry key="value.deserializer"

1212                     value="org.apache.kafka.common.serialization.StringDeserializer" />

1313             </map>

1414         </constructor-arg>

1515     </bean>

1616 

1717     <!-- 2.创建consumerFactory bean -->

1818     <bean id="consumerFactory"

1919         class="org.springframework.kafka.core.DefaultKafkaConsumerFactory" >

2020         <constructor-arg>

2121             <ref bean="consumerProperties" />

2222         </constructor-arg>

2323     </bean>

2424 

2525     <!-- 3.定义消费实现类 -->

2626     <bean id="kafkaConsumerService" class="xxx.service.impl.KafkaConsumerSerivceImpl" />

2727 

2828     <!-- 4.消费者容器配置信息 -->

2929     <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">

3030         <!-- topic -->

3131         <constructor-arg name="topics">

3232             <list>

3333                 <value>${kafka.consumer.topic.credit.for.lease}</value>

3434                 <value>${loan.application.feedback.topic}</value>

3535                 <value>${templar.agreement.feedback.topic}</value>

3636                 <value>${templar.aggrement.active.feedback.topic}</value>

3737                 <value>${templar.aggrement.agreementRepaid.topic}</value>

3838                 <value>${templar.aggrement.agreementWithhold.topic}</value>

3939                 <value>${templar.aggrement.agreementRepayRemind.topic}</value>

4040             </list>

4141         </constructor-arg>

4242         <property name="messageListener" ref="kafkaConsumerService" />

4343     </bean>

4444     <!-- 5.消费者并发消息监听容器，执行doStart()方法 -->

4545     <bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart" >

4646         <constructor-arg ref="consumerFactory" />

4747         <constructor-arg ref="containerProperties" />

4848         <property name="concurrency" value="${concurrency}" />

4949     </bean>

50

1
2
3
4
11.consumerProperties-》consumerFactory  载入配置构造消费者工厂  

22.messageListener-》containerProperties  载入容器配置（topics）  

33.consumerFactory+containerProperties-》messageListenerContainer 容器配置（topics）+消息监听器，构造一个并发消息监听容器，并执行初始化方法doStart

4

2.2消费kafka消息

方案1:直接实现
MessageListener接口，复写
onMessage方法，实现自定义消费业务逻辑。

1
2
3
4
5
6
7
8
9
10
11
12
11 public class KafkaConsumerSerivceImpl implements MessageListener<String, String> {

2 2     @Override

3 3     public void onMessage(ConsumerRecord<String, String> data) {

4 4         //根据不同主题，消费

5 5         if("主题1".equals(data.topic())){

6 6             //逻辑1

7 7         }else if("主题2".equals(data.topic())){

8 8             //逻辑2

9 9         }

1010     }

1111 }

12

方案2：使用
@KafkaListener注解，并设置topic,支持SPEL表达式。这样方便拆分多个不同topic处理不同业务逻辑。（特别是有自己的事务的时候，尤其方便）

1
2
3
4
5
6
7
8
9
11 import org.springframework.kafka.annotation.KafkaListener;

22 

33 public class KafkaConsumerSerivceImpl {

44     @KafkaListener(topics = "${templar.aggrement.agreementWithhold.topic}")

55     void templarAgreementNoticewithhold(ConsumerRecord<String, String> data){

66        //消费业务逻辑

77     }

88 }

9

三、总结

本章我们实现了一个简单的kafka生产、消费消息的实践。到这里我们已经会基本使用kafka了。是不是很简单…

下一章，我们从源码角度来深入分析spring-kafka。