Spring Integration Kafka XML 配置示例

3. 消费者配置

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

    <!-- 消费配置 -->
    <bean id="consumerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="192.168.90.200:9092"/>
                <entry key="group.id" value="0"/>
                <entry key="enable.auto.commit" value="true"/>
                <entry key="auto.commit.interval.ms" value="1000"/>
                <entry key="session.timeout.ms" value="15000"/>
                <entry key="key.deserializer" value="org.apache.kafka.common.serialization.IntegerDeserializer"/>
                <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            </map>
        </constructor-arg>
    </bean>

    <!-- 创建 consumerFactory bean -->
    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg>
            <ref bean="consumerProperties"/>
        </constructor-arg>
    </bean>

    <!-- 消费通道 -->
    <int:channel id="inputFromKafka">
        <int:queue/>
    </int:channel>

    <int-kafka:message-driven-channel-adapter auto-startup="true" channel="inputFromKafka" listener-container="container1"/>

    <bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
        <constructor-arg index="0" ref="consumerFactory"/>
        <constructor-arg index="1" ref="containerProperties"/>
    </bean>

    <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
        <constructor-arg value="Xuesong_Topic"/>
    </bean>

</beans>

4. 测试

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:applicationContext-kafka.xml"})
public class Test {

    @Autowired
    @Qualifier("inputToKafka")
    MessageChannel messageChannel;

    @Autowired
    @Qualifier("inputFromKafka")
    PollableChannel pollableChannel;

    @Test
    public void sendMsg() throws Exception {
        for (int i = 0; i < 15; i++) {
            Message<String> message = new GenericMessage<String>("test-------------" + (i + 100));
            boolean flag = messageChannel.send(message);
            System.out.println(flag + "=============" + (i + 100));
        }

        Message<?> received = pollableChannel.receive(10000);
        while (received != null) {
            System.out.println("|||" + received);
            received = pollableChannel.receive(10000);
        }
    }
}