java - 如何中断或取消 Spring Integration Java DSL 流程？

Question

考虑实现一个功能，其中 JdbcRepositoryHandler(实现 MessageHandler)可能会监听外部事件(例如，CancelRunEvent)。

我想我会使用 Spring 的 ApplicationEvent 支持通过 REST Controller 端点发布事件。我想我应该让上述处理程序实现 ApplicationListener 来监听特定事件？

问题是:如果处理程序充满了它需要处理的消息，我将如何发出终止所有后续消息的信号，这些消息可能已经从上游发出，例如来自 FileSplitter?

虽然我可以很容易地构造一个在调用负责的方法之前要检查的条件，例如，对于持久性操作(基于从 CancelRunEvent 接收到的某些状态)，我怎么能中断完全流动？

为了便于说明，想象一下这样的流程:

@Bean
protected IntegrationFlow s3ChannelFlow() {
    // @see http://docs.spring.io/spring-integration/reference/html/files.html#file-reading
    // @formatter:off
    return IntegrationFlows
        .from(s3Channel())
        .enrichHeaders(h -> h.headerFunction(RunStats.FILE_TOKEN, f -> UUID.randomUUID().toString()))
        .channel(runStatsChannel())
        .transform(new FileToInputStreamTransformer())
        .split(new FileSplitter())
        .transform(new JsonToObjectViaTypeHeaderTransformer(new Jackson2JsonObjectMapper(objectMapper), typeSupport))
        .publishSubscribeChannel(p -> p.subscribe(persistenceSubFlow()))
        .get();
    // @formatter:on
}

@Bean
protected IntegrationFlow persistenceSubFlow() {
    // @formatter:off
    return f -> f
            // @see http://docs.spring.io/spring-integration/reference/html/messaging-routing-chapter.html#agg-and-group-to
            .aggregate(a -> a
                    .correlationStrategy(new HeaderAttributeCorrelationStrategy(RunStats.FILE_TOKEN))
                    .releaseStrategy(new MessageCountReleaseStrategy(persistenceBatchSize))
                    .sendPartialResultOnExpiry(true)
                    .expireGroupsUponCompletion(true)
                    .groupTimeoutExpression(persistenceBatchReleaseTimeoutMillis)
            )
            .handle(new JdbcRepositoryHandler(typeSupport, metricRegistry, runStatsRepository));
    // @formatter:on
}

Answer 1

目前还不完全清楚你的意思或为什么你需要 JdbcRepositoryHandler 而不是其他一些 ApplicationListener 来管理它。

您的流程正在 s3Channel() 上游的某个线程上运行。根据具体情况，您可以stop() 消息源，并且不会在当前消息(或多线程消息)之后发出新消息。

但是，您可能(很可能会)在内存中保留部分聚合，直到组超时。