一、添加依赖项
compile 'org.springframework.kafka:spring-kafka:1.2.2.RELEASE'
二、发消息(生产者)
2.1 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"
4 xsi:schemaLocation="http://www.springframework.org/schema/beans
5 http://www.springframework.org/schema/beans/spring-beans.xsd">
6
7 <bean id="producerProperties" class="java.util.HashMap">
8 <constructor-arg>
9 <map>
10 <!--kafka的服务地址,多个地址用英文逗号连接-->
11 <entry key="bootstrap.servers" value="192.168.0.10:9092,192.168.0.11:9092,192.168.0.12:9092"/>
12 <entry key="group.id" value="0"/>
13 <entry key="retries" value="10"/>
14 <entry key="batch.size" value="16384"/>
15 <entry key="linger.ms" value="1"/>
16 <entry key="buffer.memory" value="33554432"/>
17 <entry key="key.serializer" value="org.apache.kafka.common.serialization.IntegerSerializer"/>
18 <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
19 </map>
20 </constructor-arg>
21 </bean>
22
23 <bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
24 <constructor-arg>
25 <ref bean="producerProperties"/>
26 </constructor-arg>
27 </bean>
28
29 <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
30 <constructor-arg ref="producerFactory"/>
31 <constructor-arg name="autoFlush" value="true"/>
32 <!--topic名字-->
33 <property name="defaultTopic" value="dc-monitor"/>
34 </bean>
35
36 </beans>
2.2 发送代码示例
@Test
public void send() throws InterruptedException, ExecutionException, TimeoutException {
KafkaTemplate template = context.getBean(KafkaTemplate.class);
String msg = "中华人民共和国万岁!";
ListenableFuture<SendResult<String, String>> future = template.sendDefault(msg);
SendResult<String, String> result = future.get(10, TimeUnit.SECONDS);
System.out.println("发送成功=====>" + msg);
}
三、收消息(消费者)
3.1 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"
4 xsi:schemaLocation="http://www.springframework.org/schema/beans
5 http://www.springframework.org/schema/beans/spring-beans.xsd">
6
7 <bean id="consumerProperties" class="java.util.HashMap">
8 <constructor-arg>
9 <map>
10 <!--kafka的服务地址,多个地址用英文逗号连接-->
11 <entry key="bootstrap.servers" value="192.168.0.10:9092,192.168.0.11:9092,192.168.0.12:9092"/>
12 <entry key="group.id" value="0"/>
13 <entry key="enable.auto.commit" value="true"/>
14 <entry key="auto.commit.interval.ms" value="1000"/>
15 <entry key="session.timeout.ms" value="15000"/>
16 <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
17 <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
18 </map>
19 </constructor-arg>
20 </bean>
21
22 <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
23 <constructor-arg ref="consumerProperties"/>
24 </bean>
25
26 <!-- 实际执行消息消费的类 -->
27 <bean id="kafkaConsumer" class="com.cnblogs.yjmyzz.consumer.DemoKafkaConsumer"/>
28
29 <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
30 <!--topic名字-->
31 <constructor-arg value="dc-monitor"/>
32 <property name="messageListener" ref="kafkaConsumer"/>
33 </bean>
34
35 <bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
36 init-method="doStart">
37 <constructor-arg ref="consumerFactory"/>
38 <constructor-arg ref="containerProperties"/>
39 </bean>
40
41 </beans>
3.2 接收代码示例
public class DemoKafkaConsumer implements MessageListener<String, String> {
@Override
public void onMessage(ConsumerRecord<String, String> data) {
System.out.println("收到消息=====>" + data.value());
}
}