spring集成：多步多通道订户

Question

我需要实现一个由多个步骤组成的集成流程，每个步骤都可以由不同数量的处理器（插件）执行。

到目前为止我所拥有的：

<!-- gateway -->
<int:gateway default-request-channel="step/1" service-interface="ServiceGateway">
    <int:method name="send" />
</int:gateway>


<!-- plugin 1 -->
<int:publish-subscribe-channel id="step/1" apply-sequence="true" />

<int:service-activator input-channel="step/1" output-channel="step/2">
    <bean class="Transformer" />
</int:service-activator>

<int:service-activator input-channel="step/1" output-channel="step/2">
    <bean class="Transformer2" />
</int:service-activator>


<!-- plugin 2 -->
<int:publish-subscribe-channel id="step/2" apply-sequence="true" />

<int:service-activator input-channel="step/2" output-channel="end">
    <bean class="Transformer3" />
</int:service-activator>

<int:service-activator input-channel="step/2" output-channel="end">
    <bean class="HttpTransformer4" />
</int:service-activator>

<!-- aggregation -->
<int:channel id="end" />
<int:aggregator input-channel="end" />

预期的行为如下：

1. 通过网关发送第一个请求

一切正常，但结果不是预期的，我只收到2个（随机）项目，而不是4个。

我认为问题在于聚合器仅在两个项目之后触发发布，因为“step/2”通道中的“apply sequence”覆盖了“step/1”中的“apply sequence”。所以问题是：如何让聚合器等待所有消息？

提前谢谢你。

定制发布策略：

@SuppressWarnings("unchecked")
@Override
public boolean canRelease ( MessageGroup group ) {

    MessageHeaders headers = group.getOne ().getHeaders ();
    List<List<Object>> sequenceDetails = (List<List<Object>>) headers.get ( "sequenceDetails" );
    System.out.println ( sequenceDetails );

    int expectedSize = 1;
    //map message id, max group size reached (i.e. sequenceNumber==sequenceSize)
    for ( List<Object> sequenceItem : sequenceDetails ) {
        if ( sequenceItem.get ( 1 ) != sequenceItem.get ( 2 ) ) {
            System.err.println ( "--> AGG: no release check, group max not reached" );
            return false;
        }
        expectedSize *= (int) sequenceItem.get ( 2 );//multiplies the group sizes
    }

    int expectedSize2 = expectedSize * (int) headers.get ( "sequenceSize" );

    int currentSize = group.getMessages ().size () * expectedSize;
    System.err.println ( "--> AGG: " + expectedSize2 + " : " + currentSize );
    boolean canRelease = expectedSize2 == currentSize;
    if ( canRelease ) {
        System.out.println ( "ok" );
    }
    return canRelease;
}

打印出来：

[[7099b583-55d4-87d3-4502-993f05bfb388,1,2]]

--

[[7099b583-55d4-87d3-4502-993f05bfb388,1,2]]

--

[[7099b583-55d4-87d3-4502-993f05bfb388,2,2]]

--

[[7099b583-55d4-87d3-4502-993f05bfb388,2,2]]

--

聚合代码：

@Aggregator
public Object aggregate ( List<Message<?>> objects ) {

    List<Object> res = new ArrayList<> ();
    for ( Message<?> m : objects ) {
        res.add ( m.getPayload () );
        MessageHeaders headers2 = m.getHeaders ();
        System.out.println ( headers2.get ( "history" ) );
    }

    return res;
}

打印出来：

网关2，核心通道，（内部bean）#57018165，异步/步骤/1，核心通道，（内部bean）#57018165，异步/步骤/2，核心通道，（内部bean）#57018165，end2

网关2，核心通道，（内部bean）#57018165，异步/步骤/1，核心通道，（内部bean）#57018165，异步/步骤/2，核心通道，（内部bean）#57018165，end2

[102, 202] --

Answer 1

使用自定义发布策略。第二个pubsub将来自第一个pubsub的相关数据推送到setence细节头中的堆栈上。

编辑

问题是有两个群体；您需要在初始correlationId上进行关联。这里有一个纯SpEL解决方案；使用自定义关联/发布策略来确保数据如预期的那样（并使用getOne（）而不是迭代器）可能更安全。。。

<int:aggregator input-channel="end2"
        correlation-strategy-expression=
           "headers['sequenceDetails'][0][0]"
        release-strategy-expression=
           "size() == iterator().next().headers['sequenceSize'] * iterator().next().headers['sequenceDetails'][0][2]" />