如何在Spring集成中推迟消息的消耗

Question

我正在开发一个使用Spring Integration 5.0.1和Spring Boot 2.0.0的应用程序。RC1

目前，应用程序响应Application ationReadyEvents并运行一些可能需要一段时间才能完成的初始化代码。这不使用任何Spring集成组件。

我还有一些非常基本的集成流，使用JavaDSL编写，并在配置中声明为bean。

有什么方法可以推迟流何时开始消耗消息吗？我希望能够在初始化完成时手动启动它们。

配置ControlBus似乎是解决方案，但我不知道如何将类似的东西与其他流连接起来。

以下是流如何使用消息的示例：

IntegrationFlows.from(sourceGateway)
                .transform(Transformers.fromJson(IncomingTask.class, jsonObjectMapper))

                .handle(IncomingTask.class, (incomingTask, headers) -> {

                //stuff with the task here.

                })

                .get();

Answer 1

好的，你绝对可以在这件事上使用ControlBus。对于Java DSL，它看起来像：

@Bean
public IntegrationFlow controlBus() {
    return IntegrationFlowDefinition::controlBus;
}

要使用它，您需要：

@Autowired
@Qualifier("controlBus.input")
private MessageChannel controlBusChannel;

现在我们需要知道您的目标intgraion Flow是如何启动的。什么消耗消息。例如，我有这个：

@Bean
public IntegrationFlow fileFlow1() {
    return IntegrationFlows.from("fileFlow1Input")
            .handle(Files.outboundAdapter(tmpDir.getRoot()),
                        c -> c.id("fileWriting").autoStartup(false))
                .get();
    }

注意c.id（"fileWrit"）. autoStartup（false）。id是用于endpointbean的，它可以通过发送到Control Bus的命令来访问。autoStartup（false）意味着它不会立即消耗消息，而是只有当我们调用start（）时才会消耗消息。我是这样做的：

this.controlBusChannel.send(new GenericMessage<>("@fileWriting.start()"));

您应该在配置中确保将消息消耗推迟到您需要的时间。