Spring Integration 通过控制总线手动启动/停止通道适配器答案

【问题标题】：Spring Integration manually start/stop channel adapter via control busSpring Integration 通过控制总线手动启动/停止通道适配器
【发布时间】：2014-05-28 15:14:20
【问题描述】：

是否有手动启动/初始化通道适配器的方法？

我的 context.xml 中有两对入站/出站适配器，我想在运行时决定要开始使用其中的哪一个。

编辑：

具体场景：
我有一个客户端，可以在运行时配置为 mqtt 发布者或订阅者。
我的 context.xml 看起来像这样：

<int-mqtt:message-driven-channel-adapter 
    client-id="foo"
    auto-startup="true"
    url="tcp://192.168.97.164:1883"
    topics="testtopic/#"
    channel="writeToFile" />

<file:outbound-channel-adapter
    id="writeToFile"
    auto-startup="true"
    directory="./test/out"
    delete-source-files="false"/>

<int:transformer id="Transformer"
    ref="MessageTransformer"
    input-channel="readFromFile"
    output-channel="mqttOut"
    method="bytesFromFile" />

<bean id="MessageTransformer" class="MessageTransformer"/>

<int-mqtt:outbound-channel-adapter 
    id="mqttOut"
    client-id="foo"
    url="tcp://192.168.97.164:1883"
    auto-startup="false"
    default-qos="1"
    default-retained="true"
    default-topic="testtopic/bla"
    />

    <file:inbound-channel-adapter
    auto-startup="false" 
    id="readFromFile"
    directory="./test/in"
    filename-pattern="myFile*">
    <int:poller id="poller"
        fixed-rate="5000" />     
</file:inbound-channel-adapter>

如您所见，我有两个设置：
1.订阅者案例：读取mqtt消息->写入文件
2. Publisher案例：从目录轮询文件->通过mqtt发送

我在运行时决定要应用什么设置。

那么你能告诉我这个控制总线的东西如何适合这里吗？

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

设置autoStartup="false" 或者直接start()/stop() 他们，或者使用<control-bus/>（发送@myAdapter.start()）。

获取直接引用（autowire 等）取决于端点类型。如果是轮询端点，则注入SourcePollingChannelAdapter；消息驱动的适配器各不相同，但通常是 MessageProducerSupport 或 MessagingGatewaySupport。

编辑：

阅读control-bus here。

给入站适配器一个id 属性。

添加<control-bus input-channel="control"/>

添加<int:gateway service-interface="foo.Controller" default-request-channel="control"/>

创建网关接口

public interface Controller {

    void control(String command);

}

@Autowire 网关（或使用context.getBean(Controller.class)）。

然后，当您准备好启动适配器时，调用例如gateway.control("@mqttOut.start()").

出站适配器上不需要auto-startup="false"。

但是，对于像这样的简单用例，您可能希望改为使用 Spring 配置文件进行调查（将适配器放在配置文件中并在运行时启用配置文件。

【讨论】：

谢谢你们，但恐怕我还需要一些关于控制总线的进一步帮助。因此，我刚刚更新了问题。
我添加了有关如何在此场景中使用控制总线的详细信息。但您可能想查看 Spring 配置文件。
谢谢。这适用于第一种情况（启动 mqqt:message-driven-channel-adapter），但是当我尝试 '@readFromFile.start()' 时出现以下错误：此命令不支持方法 'start'处理器。如果使用控制总线，请考虑添加“@ManagedOperation”或“@ManagedAttribute”。
这是因为您在入站适配器上使用了隐式通道命名；当你不给通道适配器一个channel 属性时，通道会得到id readFromFile 并且适配器被称为readFromFile.adapter。您正在尝试 start() 通道，这是不允许的。因此，要么在适配器上使用显式 id 和通道，要么使用 @'readFromFile.adapter'.start()。
谢谢。我决定暂时使用 Spring 配置文件。很棒的小费。我还有一个问题：网关/控制总线和我的入站适配器之间的链接到底在哪里？我承认，它正在工作，但我仍然无法弄清楚如何！

【解决方案2】：

为此，您需要先将通道适配器自动启动属性设置为 false auto-startup="false"，然后使用控制总线启动/停止适配器

参见此处控制总线示例 - https://github.com/spring-projects/spring-integration-samples/tree/master/basic/control-bus

【讨论】：

谢谢你们，但恐怕我还需要一些关于控制总线的进一步帮助。因此，我刚刚更新了问题。

【解决方案3】：

我正在寻找使用 spring 集成 Java DSL 的相同示例，但没有找到任何东西，所以我创建了自己的示例。配置起来非常简单。

@Bean
public IntegrationFlow controlBus() {
    return IntegrationFlows.from(controlChannel())
            .controlBus()
            .get();
}

@Bean
public MessageChannel controlChannel() {
    return MessageChannels.direct().get();
}

停止它：

controlChannel.send(new GenericMessage<>("@myInboundAdapter.stop()"));

https://github.com/CauchyPeano/sftp-poller-control-bus

【讨论】：