一、添加依賴項

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 發送代碼示例

@Testpublic 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> {@Overridepublic void onMessage(ConsumerRecord<String, String> data) {System.out.println("收到消息=====>" + data.value());} }

　

總結

以上是生活随笔為你收集整理的spring集成kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。