Spring整合Kafka

maven

 <!-- kafka与spring整合 -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.0.4.RELEASE</version>
        </dependency>

生产者

package com.ydfind.kafka.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;

public class KfkSpringProducer {
    public static void main(String[] args) {
        ApplicationContext context = new ClassPathXmlApplicationContext("spring-kafka.xml");
        // KafkaTemplate是spring用来发送消息的模板类
        KafkaTemplate<String, String> kafkaTemplate = (KafkaTemplate) context.getBean("kafkaTemplate");
        kafkaTemplate.send("testGroup2", "我的测试消息1");
        System.out.println("have send msg = 我的测试消息1");
        kafkaTemplate.send("testGroup2", "我的测试消息2");
        System.out.println("have send msg = 我的测试消息2");
        kafkaTemplate.send("testGroup2", "我的测试消息3");
        System.out.println("have send msg = 我的测试消息3");
    }
}

消费者

package com.ydfind.kafka.spring;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;
public class KfkConsumerListener implements MessageListener<String, String> {
    /**
     * 监听器自动执行该方法
     * 消费消息
     * 自动提交offset
     *
     * @param record
     */
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        System.out.printf("partition = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
    }
}

配置文件spring-kafka.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">

    <!--生产者配置-->
    <bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
        <constructor-arg name="configs">
            <map>
                <entry key="bootstrap.servers" value="localhost:9092,localhost:9093,localhost:9094"/>
                <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                <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="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg ref="producerFactory"/>
        <constructor-arg name="autoFlush" value="true"/>
    </bean>

    <!--消费者配置 -->
    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg name="configs">
            <map>
                <entry key="bootstrap.servers" value="localhost:9092,localhost:9093,localhost:9094"/>
                <entry key="group.id" value="testGroup2"/>
                <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="consumerListener" class="com.ydfind.kafka.spring.KfkConsumerListener"/>
    <bean id="containerProperties_example" class="org.springframework.kafka.listener.config.ContainerProperties">
        <constructor-arg value="testGroup2"/>
        <property name="messageListener" ref="consumerListener"/>
    </bean>
    <bean id="messageListenerContainer_example" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
        <constructor-arg ref="consumerFactory"/>
        <constructor-arg ref="containerProperties_example"/>
    </bean>
</beans>