是否可以将 Spring Integration 用于带有 ExecutorChannel 的批处理或带有 Task Executor 的 Pollers答案

【问题标题】：Is it possible to use Spring Integration for batch processes with ExecutorChannel or Pollers with Task Executors是否可以将 Spring Integration 用于带有 ExecutorChannel 的批处理或带有 Task Executor 的 Pollers
【发布时间】：2020-03-27 21:58:10
【问题描述】：

我正在尝试将 Spring Integration 用于批处理。某些步骤很耗时，因此如果 QueueChannel 有多个消费者，每个消费者都在单独的线程上运行，这将受益。

这种方法的问题在于，在所有消息都被消耗后，没有干净的方法来关闭应用程序。我曾尝试使用控制总线并关闭任务执行器，但这只有在您能猜到所有消息将在什么时间被消耗并且没有任何消息处于飞行状态时才有效，这是不可能的。

对于批处理是否有一种干净的方法来执行此操作，或者这只是使用 Spring Integration 的错误用例？

编辑：

本质上，如果我有一种方法可以发送一条特殊消息，该消息表示生命周期事件之一，例如启动或停止，它会自动通过所有 Spring 集成组件进行。这样可以保证停止消息到达最后，并且如果有办法在停止消息到达时触发 生命周期 感知 bean 上的 shutdown()。

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

在所有消息都被消费完后关闭应用程序

如果您知道一开始的消息数量，您可以拥有一个AtomicInteger bean，并在每次处理消息时递增它。另外，您有一个入站通道适配器来轮询 AtomicInteger 的当前状态并决定关闭您的应用程序。

或者，您可以使用聚合器收集所有消息的结果，并在聚合器之后关闭下游流中的应用。

【讨论】：

如果无法预测有多少消息可以到达并且正在处理的消息数量存在巨大差异，那么有可能吗？我不想猜测可能有多少条消息。本质上，如果我有办法发送一条特殊消息来表示生命周期事件之一，例如启动或停止，它会自动通过所有 Spring 集成组件进行。这样可以保证停止消息到达最后，并且如果有一种方法可以在停止消息到达时触发生命周期感知 bean 上的 shutdown()。
消息来源是什么？它不能无限到你根本没有理由停止你的应用程序吗？否则该来源不能为您发送“停止”消息吗？如果一段时间没有活动，可以自行停止应用程序。例如，请参阅SimpleActiveIdleMessageSourceAdvice。