SpringXD：自动启动=false在Kafka消息驱动通道适配器上不起作用

Question

我使用的是< code>SpringXD，我的配置如下:

• Spring集成kafka 2.1.0.释放
• kafka客户端0.10.0.1
• Kafka0.10.x.x
• Spring-xd-1.3.1.释放

我的 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" xmlns:int="http://www.springframework.org/schema/integration"
    xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
    xmlns:task="http://www.springframework.org/schema/task"
    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-3.0.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">


    <int:channel id="input" />
    <int:channel id="output" />

    <int:control-bus input-channel="input" />

    <int-kafka:message-driven-channel-adapter
        id="kafka-inbound-channel-adapter-testing" listener-container="container1"
        auto-startup="false" phase="100" send-timeout="5000"
        channel="output" mode="record"
        message-converter="messageConverter" />

    <bean id="messageConverter" class="org.springframework.kafka.support.converter.MessagingMessageConverter" />

    <!--Consumer -->
    <bean id="container1"
        class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
        <constructor-arg>
            <bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
                <constructor-arg>
                    <map>
                        <entry key="bootstrap.servers" value="localhost:9092" />
                        <entry key="enable.auto.commit" value="false" />
                        <entry key="auto.commit.interval.ms" value="100" />
                        <entry key="session.timeout.ms" value="15000" />
                        <entry key="max.poll.records" value="3" />
                        <entry key="group.id" value="bridge-stream-testing" />
                        <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>
        </constructor-arg>

        <constructor-arg>
            <bean class="org.springframework.kafka.listener.config.ContainerProperties">
                <constructor-arg name="topics" value="testing-topic" />
            </bean>
        </constructor-arg>
    </bean>

</beans>

这是我用来启动/停止频道的Java类：

package com.kafka.source.logic;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

@Configuration
@EnableScheduling
@ImportResource("classpath:/config/kafka-source-context.xml")
public class KafkaSourceRetry {

    @Autowired
    MessageChannel input;

    @Scheduled(cron="*/50 * * * * *")
    void startAdapter(){
        //CODE COMMENTED OUT TO MAKE SURE THE ADAPTER IS NOT BEING STARTED
        //EVEN IF I UNCOMMENT THE CODE, THE 50 secs defined related to the cron are not respected.
        //That is, if I send a message to the topic, it is inmediately consumed
        //input.send(new GenericMessage<String>("@kafka-inbound-channel-adapter-testing.start()"));
    }
}

然后我创建了一个基本流来检查我发送到主题的一些消息是否通过

stream create --name bridgeStream --definition "kafkaSourceLatestApi_v2|bridge|file" --deploy

我检查了创建的文件，它包含我发送到 Kafka 主题的所有消息：

hola _ que _ tal que _ bonito bridgestream . out(END)

同样在日志中，我发现了这个：

2017-04-10T22：37：06-0300 1.3.1.发布信息 部署路径儿童缓存-0 支持。默认生命周期处理器 - 在第 0 阶段启动 bean 2017-04-10T22：37：06-0300 1.3.1.发布调试部署路径子缓存-0 支持。默认生命周期处理器 - 启动 bean 'container1' 的类型 [class org.springframework.kafka.listener.KafkaMessageListenerContainer] 2017-04-10T22：37：06-0300 1.3.1.发布调试部署路径儿童缓存-0 支持。默认生命周期处理器 - 已成功启动 bean 'container1' 2017-04-10T22：37：06-0300 1.3.1.发布信息 部署路径儿童缓存-0 支持。默认生命周期处理器 - 在第 100 阶段启动 bean 2017-04-10T22：37：06-0300 1.3.1.发布调试部署路径儿童缓存-0 支持。默认生命周期处理器 - 启动 bean 'kafka-inbound-channel-adapter-testing' 类型 [class org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter] 2017-04-10T22：37：06-0300 1.3.1.发布信息 部署路径儿童缓存-0 入站。KafkaMessageDrivenChannelAdapter - 已启动 kafka-inbound-channel-adapter-test 2017-04-10T22：37：06-0300 1.3.1.发布调试部署PathChildrenCache-0支持。默认生命周期处理器 - 已成功启动 bean 'kafka-inbound-channel-adapter-testing'

我的问题是:为什么频道会自动启动？

Answer 1

它是这样设计的;所有模块的自动启动都设置为 false，因此它们不会乱序启动;部署流时，各个模块将从右到左部署和启动。

部署/取消部署是启动/停止流的方式。

请参阅模块部署程序。