java - Spring Integration Aggregator - 丢失消息

Question

我想收集一些消息(比如 10 条)并将它们作为列表传递给服务激活器，而不是一条一条地传递。

上下文:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns=...>

    <int:channel id="ch.http.in"/>
    <int:channel id="ch.http.trans"/>
    <int:channel id="ch.http.aggr"/>
    <int-http:inbound-channel-adapter path="test" channel="ch.http.in"/>

    <int:map-to-object-transformer input-channel="ch.http.in" output-channel="ch.http.trans" type="demo.Req"/>
    <int:aggregator 
        input-channel="ch.http.trans" 
        output-channel="ch.http.aggr"
        release-strategy-expression="size() == 10"
        correlation-strategy-expression="headers['id']"
        ref="aggr" method="add"/>
    <int:service-activator ref="srv" method="httpTest" input-channel="ch.http.aggr"/>

    <bean id="srv" class="demo.IntService"/>
    <bean id="aggr" class="demo.HttpAggregator"/>
</beans>

聚合器:

public class HttpAggregator{
    public List<Req> add(List<Req> reqs) {
        System.out.println(reqs);
        return reqs;
      }
}

服务:

public class IntService {
    public void httpTest(Req msg){
        System.out.println(msg);
    }
}

Req 只是一个 POJO。

问题是从未调用聚合器方法。如果没有聚合器，消息将毫无问题地传递到服务激活器。使用 Spring Integration 3.0.2.RELEASE(Spring Boot 1.0.2.RELEASE)

编辑:当我将 correlation-strategy-expression="headers['id']" 更改为 correlation-strategy-expression="payload.id"(Req 对象具有属性 id ) 当我为每个 block 传递不同的 id 时它会起作用(例如前 10 个 id=1；接下来的 10 个 id=1...)看起来这就是相关策略的工作原理。我怎样才能绕过它？我只想限制聚合列表的大小。

Answer 1

对；您必须关联某些东西；使用 headers['id'] 最终会得到很多永远不会满足发布策略的 1 组项目。

对于像您这样的简单用例，关联文字 - 例如correlation-expression="'foo'" 并设置 expire-groups-on-completion="true"。这会在发布后重置组，因此可以在下一条消息上启动一个新组(具有相同的相关 ID)。

如果您想在超时后释放部分组，您将需要一个MessageGroupStoreReaper。或者，如果您可以升级到 4.0.x，聚合器现在有一个 group-timeout(或 group-timeout-expression)。