Spring集成并行拆分答案

【问题标题】：Spring integation parallel splitSpring集成并行拆分
【发布时间】：2023-02-24 01:58:36
【问题描述】：

我有这样的代码。是否可以控制第一次拆分的顺序？

` @Bean
  public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(50);
    return executor;
  }


  @Bean
  public IntegrationFlow firstFlow() {
    return IntegrationFlows.from("firstChannel")
        .split()
        .channel("inputChannel")
        .get();
  }
  
  @Bean
  public IntegrationFlow inputFlow() {
    return IntegrationFlows.from("inputChannel")
        .channel(MessageChannels.executor(taskExecutor()))
        .split()
        .handle(this::mapping)
        .aggregate()
        .channel("aggregateChannel")
        .get();
  }

  @Bean
  public IntegrationFlow aggregateFlow() {
    return IntegrationFlows.from("aggregateChannel")
        .aggregate()
        .get();
  }`

我想对方法“映射”进行异步处理，但要开始处理来自第一个拆分的第二条消息，并且仅当第一条消息出现在 aggregateChannel 中时才发送到 inputChannel

【问题讨论】：

我是否正确理解您想要按顺序拆分原始消息，但并行处理它们的项目？那么，只有当第一个完全处理后，您才会开始处理下一个，即使它的项目是并行处理的？
是的，这就是我想做的

标签： spring spring-integration

【解决方案1】：

因此，这是可能解决方案的单元测试：

@SpringJUnitConfig
public class So75547720Tests {

    @Autowired
    BeanFactory beanFactory;

    @Test
    void sequentialSplitButSubSplitParallel() {
        List<String> firstList = List.of("1", "2", "3", "4");
        List<String> secondList = List.of("5", "6", "7", "8");
        List<List<String>> testData = List.of(firstList, secondList);

        MessagingTemplate messagingTemplate = new MessagingTemplate();
        messagingTemplate.setBeanFactory(this.beanFactory);

        List<List<String>> result = messagingTemplate.convertSendAndReceive("firstChannel", testData, List.class);
        assertThat(result).isNotNull().hasSize(2);
        assertThat(result.get(0)).hasSameElementsAs(firstList);
        assertThat(result.get(1)).hasSameElementsAs(secondList);
        System.out.println(result);
    }

    @Configuration
    @EnableIntegration
    public static class TestConfiguration {

        @Bean
        public TaskExecutor taskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(5);
            return executor;
        }


        @Bean
        public IntegrationFlow firstFlow() {
            return IntegrationFlow.from("firstChannel")
                    .split()
                    .channel("inputChannel")
                    .get();
        }

        @Bean
        public IntegrationFlow inputFlow() {
            return IntegrationFlow.from("inputChannel")
                    .gateway(subFlow -> subFlow
                            .split()
                            .channel(MessageChannels.executor(taskExecutor()))
                            .handle(this::mapping)
                            .aggregate())
                    .channel("aggregateChannel")
                    .get();
        }

        @Bean
        public IntegrationFlow aggregateFlow() {
            return IntegrationFlow.from("aggregateChannel")
                    .aggregate()
                    .get();
        }

        private String mapping(String payload, Map<String, ?> headers) {
            System.out.println("Handling thread: " + Thread.currentThread().getName() + " for: " + payload);
            return payload.toUpperCase();
        }

    }

}

第一个 split() 按顺序向 inputChannel 发出项目。然后我们使用 gateway 作为子流。此网关将等待回复以将其推送到下一个aggregateChannel。有趣的部分确实是在我们使用第二个拆分器的子流中，它确实根据 Executor 通道并行发出项目。内部聚合器在收集当前拆分的所有项目之前不会发出。只有在那之后，我们才会从顶级拆分中获取下一个项目。

测试的结果可能是这样的：

Handling thread: taskExecutor-2 for: 2
Handling thread: taskExecutor-1 for: 1
Handling thread: taskExecutor-3 for: 3
Handling thread: taskExecutor-4 for: 4
Handling thread: taskExecutor-2 for: 6
Handling thread: taskExecutor-5 for: 5
Handling thread: taskExecutor-3 for: 7
Handling thread: taskExecutor-1 for: 8
[[2, 3, 1, 4], [6, 5, 7, 8]]

【讨论】：