Spring Integration Aggregator - 丢失消息答案

【问题标题】：Spring Integration Aggregator - lost messagesSpring Integration Aggregator - 丢失消息
【发布时间】：2014-06-10 18:54:57
【问题描述】：

我想收集一些消息（比如说 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）时，它在我为每个块传递不同的 id 时起作用（例如，前 10 个的 id=1；接下来的 10 个为 2...）看起来就是这样关联策略是如何工作的。我怎样才能通过它？我只是想限制聚合列表的大小。

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

对；你必须与某事相关联；使用 headers['id'] 最终会得到很多一组 1 项，这些项永远不会满足发布策略。

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

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

【讨论】：