Spring 将 Scatter-Gather 模式与 JMS 传输集成答案

【问题标题】：Spring integration Scatter-Gather pattern with JMS transportSpring 将 Scatter-Gather 模式与 JMS 传输集成
【发布时间】：2018-01-20 01:57:04
【问题描述】：

我需要实现以下架构：

我有必须使用 JMS 发送到系统（某些外部应用程序）的数据。
根据您需要发送的数据只发送到必要的系统（例如，如果系统数量为 4，那么您可以从 1 发送到 4）
需要等待消息发送到的系统的响应，收到所有响应后，需要处理接收到的数据（或至少处理一次超时）
相关 ID 包含在传出和传入 JMS 消息的标头中
每个新的此类进程都可以异步并行启动

现在我只在 Spring JMS 的帮助下实现了它。我手动同步线程，也手动管理线程池。

发送消息的系统的相关ID和信息被存储为状态，并在收到新消息等后更新。

但我想简化逻辑并使用 Spring 集成 Java DSL、分散聚集模式（这只是我的情况）和其他有用的 Spring 特性。

你能帮我展示一个如何在 Spring-integration/IntregrationFlow 的帮助下实现这种架构的例子吗？

【问题讨论】：

标签： spring jms spring-integration enterprise-integration

【解决方案1】：

以下是我们测试用例的一些示例：

    @Bean
    public IntegrationFlow scatterGatherFlow() {
        return f -> f
                .scatterGather(scatterer -> scatterer
                                .applySequence(true)
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10))
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10))
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)),
                        gatherer -> gatherer
                                .releaseStrategy(group ->
                                        group.size() == 3 ||
                                                group.getMessages()
                                                        .stream()
                                                        .anyMatch(m -> (Double) m.getPayload() > 5)),
                        scatterGather -> scatterGather
                                .gatherTimeout(10_000));
    }

所以，有部分：

scatterer - 向收件人发送消息。在您的情况下，所有这些 JMS 服务。那可以是scatterChannel。通常是PublishSubscribeChannel，所以Scatter-Gather 可能事先不知道订阅者。
gatherer - 好吧，它只是一个 aggregator 及其所有可能的选项。
scatterGather - 只是为了方便 ScatterGatherHandler 的直接属性和常用端点选项。

【讨论】：