Spring集成聚合器时间到期 - 问题答案

【问题标题】：Spring integration aggregator time expire - issueSpring集成聚合器时间到期 - 问题
【发布时间】：2015-10-21 10:32:15
【问题描述】：

下面的代码在继续出站通道之前接受 2 条消息。

<bean id="timeout"
    class="org.springframework.integration.aggregator.TimeoutCountSequenceSizeReleaseStrategy">
    <constructor-arg name="threshold" value="2" />
    <constructor-arg name="timeout" value="7000" />
</bean> 

<int:aggregator ref="updateCreate" input-channel="filteredAIPOutput" 
        method="handleMessage" release-strategy="releaseStrategyBean" release-strategy-method="timeout">
</int:aggregator>

我的用例是整理所有消息 10 分钟并将其发送到出站通道。不是基于上面显示的消息数。为了实现这个基于时间的功能，使用下面的代码：

<int:aggregator ref="updateCreate" input-channel="filteredAIPOutput" 
        method="handleMessage" 
                output-channel="outputappendFilenameinHeader" >
</int:aggregator>

<bean id="updateCreate" class="helper.UpdateCreateHelper"/>

我传递了 10 条消息，PojoDateStrategyHelper canRelease 方法被调用了 10 次。

尝试使用时差逻辑实现 PojoDateStrategyHelper，它按预期工作。 10 分钟后调用 UpdateCreateHelper 类，但它只收到 1 条消息（最后一条消息）。其余 9 条消息在任何地方都没有看到。我在这里做错什么了吗？消息未整理。

我怀疑 SI 中应该有一些内置的东西，可以实现这一点，如果我传递 10 分钟作为参数，一旦它超过 10 分钟的时间，它应该将所有消息传递到出站通道。

这是我的 UpdateCreateHelper.java 代码：

public Message<?> handleMessage(List<Message<?>> flights){

    LOGGER.debug("orderItems list ::"+flights.size()); // this is always printing 1

    MessageBuilder<?> messageWithHeader = MessageBuilder.withPayload(flights.get(0).getPayload().toString());
    messageWithHeader.setHeader("ftp_filename", "");

    return messageWithHeader.build();
}

@CorrelationStrategy
public String correlateBy(@Header("id") String id) {
    return id;
}
@ReleaseStrategy
public boolean canRelease(List<Message<?>> flights) {
    LOGGER.debug("inside canRelease ::"+flights.size()); // This is called for each and every message
    return compareTime(date.getTime(), new Date().getTime());
}

我是 SI (v3.x) 的新手，我搜索了很多时间限制相关的聚合器，找不到任何有用的来源，请建议。

谢谢！

【问题讨论】：

标签： spring spring-integration

【解决方案1】：

打开 DEBUG 日志，看看为什么你只看到一条消息。

我怀疑 SI 中应该有一些内置的东西可以实现这一点，...

在 4.0 版之前（以及默认情况下，之后），聚合器是一个完全被动的组件；发布策略只有在有新消息到达时才会咨询。

4.0 added group timeout capabilities 从而可以在超时后释放（或丢弃）部分组。

但是，对于任何版本，您都可以配置 MessageGroupStoreReaper 以在超时后释放部分完整的组。见the documentation。

【讨论】：

【解决方案2】：

private String correlationId = date.toString();

@CorrelationStrategy
public String correlateBy(Message<?> message) {
    **// Return the correlation ID which is the timestamp the current window started (all messages should have the same correlation id)**
    return "same";
}

之前我返回的是 Header Id，它与 Message 到 Message 不同。我希望这个解决方案可以帮助一些人。忽略这么小的概念，我浪费了将近 2 天时间。

【讨论】：