分解 DSL 集成流

Question

我一直在玩 Spring Integration (SI) DSL。我有一个定义了以下异步网关的 Rest 服务：

@MessagingGateway
public interface Provision {
    @Async
    @Gateway(requestChannel = "provision.input")
    ListenableFuture<List<ResultDto>> provision(List<ItemsDto> stuff);
}

在逐行演练中，我有以下示例 IntegrationFlow。

@Bean
public IntegrationFlow provision() {
    return f -> f
            .split(ArrayList.class, List::toArray)
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .<ItemsDto, String>route(ItemsDto::getType, m -> m
                            .subFlowMapping("IPTV", sf -> sf
                                            .<ItemsDto, String>route(ItemsDto::getAction, m2 -> m2
                                                    .subFlowMapping("OPEN", sf2 -> sf2
                                                            .handle((p, h) -> iptvService.open((ItemsDto) p))))
                            )
            )
            .aggregate();
}

如您所见，我有几层路由。我需要把事情分解一下。我已经尝试了几件不起作用的事情（在这里我没有得到响应......线程不等待）：

@Bean(name = "routerInput")
private MessageChannel routerInput() {
    return MessageChannels.direct().get();
}

@Bean
public IntegrationFlow provision() {
    return f -> f
            .split(ArrayList.class, List::toArray)
            .channel(c -> c.executor(Executors.newCachedThreadPool()))
            .<ItemsDto, String>route(ItemsDto::getType, m ->
                            m.subFlowMapping("IPTV", sf -> sf.channel("routerInput"))
            )
            .aggregate();
}

@Bean
public IntegrationFlow action() {
    return IntegrationFlows.from("routerInput")
            .<ItemsDto, String>route(ItemsDto::getAction, m -> m
                    .subFlowMapping("OPEN", sf -> sf
                            .handle(p -> iptvService.open((ItemsDto) p.getPayload())))).get();
}

我显然在概念上遗漏了一些东西 :) 有人可以提供“如何以及为什么”的意见吗？

我有一个需要拆分的项目列表，按“类型”路由，然后按“动作”路由，最后聚合（包含处理程序的响应）。每个处理的项目需要并行处理。

提前致谢

更新： 根据 Artem 的建议，我删除了所有异步内容。我把它修剪到几乎没有......

@Bean(name = "routerInput")
private MessageChannel routerInput() {
    return MessageChannels.direct().get();
}

@Bean
public IntegrationFlow provision() {
    return f -> f
            .split()
            .<ItemDto, String>route(ItemDto::getType, m ->
                    m.subFlowMapping("IPTV", sf -> sf.channel("routerInput")))
            .aggregate();
}

@Bean
public IntegrationFlow action() {
    return IntegrationFlows.from("routerInput")
            .<ItemDto, String>route(ItemDto::getAction, m -> m
                    .subFlowMapping("OPEN", sf -> sf
                            .handle((p, h) -> iptvService.open((ItemDto) p)))).get();
}

我让它通过改变来响应

.handle(p ->

到这里

.handle((p, h) ->

所以它至少会响应，但它不会聚合拆分的 3 个测试项目。输出由 1 项组成。我需要使用流收集吗？发布政策？这应该没问题吧？

Answer 1

如果您想将其拆分，使用channelMapping 可能比使用subflowMapping 更简单...

    @Bean
    public IntegrationFlow typeRoute() {
        return IntegrationFlows.from(foo())
                .split()
                .<ItemsDto, String>route(ItemsDto::getType, m -> m
                        .channelMapping("foo", "channel1")
                        .channelMapping("bar", "channel2"))
                .get();
    }

    @Bean
    public IntegrationFlow fooActionRoute() {
        return IntegrationFlows.from(channel1())
                .<ItemsDto, String>route(ItemsDto::getAction, m -> m
                        .channelMapping("foo", "channel3")
                        .channelMapping("bar", "channel4"))
                .get();
    }

    @Bean
    public IntegrationFlow barActionRoute() {
        return IntegrationFlows.from(channel1())
                .<ItemsDto, String>route(ItemsDto::getAction, m -> m
                        .channelMapping("foo", "channel5")
                        .channelMapping("bar", "channel6"))
                .get();
    }

    @Bean
    public IntegrationFlow fooFooHandle() {
        return IntegrationFlows.from(channel3())
                // handle
                .channel(aggChannel())
                .get();
    }

为其他选项创建流程并汇总每个结果：

    // fooBarHandle(), barFooHandle(), barBarHandle()


    @Bean IntegrationFlow agg() {
        return IntegrationFlows.from(aggChannel())
                .aggregate()
                .get();
    }

使用ExecutorChannels...处理并行度...

    @Bean
    public MessageChannel channel1() {
        return new ExecutorChannel(exec());
    }

    @Bean
    public MessageChannel channel2() {
        return new ExecutorChannel(exec());
    }

    @Bean
    public MessageChannel channel3() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel channel4() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel channel5() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel channel6() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel aggChannel() {
        return new DirectChannel();
    }

Answer 2

我看不出您的配置有什么大问题，它确实应该可以工作。

想要我不喜欢的有：

1. 如果您使用async Gateway (ListenableFuture<List<ResultDto>>)，则不需要@Async 注释，因为它已经在 Gateway 合约中。

2. 如果您的payload 已经是List，则无需转换List::toArray。不带参数使用.split() 就足够了。

这是设计风格。

我还不确定那里有什么问题，但您是否介意让您的整个流程同步并在此处与我们分享流程的DEBUG 并指出您在哪里看到问题。

Answer 3

将聚合移动到“动作”Bean 并且它起作用了。 感谢您的耐心:)