Spring Integration Aggregator - ReleaseStrategy - 释放资源

Question

这是一个关于如何释放 ReleaseStrategy 中使用的资源的一般问题。 例如，我指的是here 提交的响应 - 我想知道如果出现问题，清理示例中使用的地图的推荐策略是什么：

private final Map<Object, AtomicInteger>() map = new HashMap<>();

用例：收到了 1000 条组消息中的 990 条（或在所引用的示例中收到了 10 条中的 9 条），但随后超时（我们对部分结果不感兴趣）...您将如何清理地图值对于这个特定的 groupId 或一般来说：释放 ReleaseStrategy 中使用的任何其他资源的最佳方法是什么？

我正在考虑一些即将到期的地图，或者可能从外部触发 bean cleanUp 方法......但也许有一些建议的方法来处理这些情况？

也许我不知道在 groupTimeout 的情况下 ReleaseStrategy 是如何工作的，所以我有以下示例配置：

group-timeout="1000" discard-channel="discardChannel" 
end-partial-result-on-expiry="false" release-strategy="xxx"

假设我有一组由拆分器产生的 10 条消息，每条消息实际上都是线程池执行器要完成的一些工作。因此，如果作业执行池中的任何一项（比如说第二个作业）花费的时间超过了指定的组超时时间，则此特定消息将被发送到 discardChannel 但也会调用 ReleaseStrategy，对吗？

但是在这种情况下，我仍然可以在该丢弃的消息中更新 MessageGroup 吗？我假设不是 - 所以 canRelease 方法调用可以大于 group.getMessages().size()，对吧？

现在发生了什么？在超时过去之前收到的这条消息是什么？这是我仍然可以决定是否要发布 MessageGroup 中的这一条消息的情况吗？ 剩下的 8 条消息（作业）是什么，在丢弃组中的第二条消息后它们是否属于新的 groupId（我假设 expire-groups-upon-timeout 默认设置为 true），所以这 8 条消息得到一个新的 groupId 并且也被发送到 ReleaseStrategy？

另一个问题是当 canRelease 方法只返回 false 时会发生什么（具有上述配置） - 我注意到虽然我得到了如下信息：

Expiring MessageGroup with correlationKey[36293c73-6f66-4a1a-8824-9d5268e07081]
Expiring MessageGroup with correlationKey[36293c73-6f66-4a1a-8824-9d5268e07081]

在这种情况下，消息是否“被返回”到 ServiceActivator（作业执行器池 bean）？所以在这种情况下它会一次又一次地执行，所以我无法退出工作流程？

Answer 1

通常，我会在地图内容中包含某种时间戳，并安排一些任务每隔一段时间进行清理。

编辑

每当有新消息到达并已添加到组中时，都会调用发布策略。如果组没有被释放并且超时，释放策略会再调用一次，从而决定是否释放组。如果是这样，这与不释放 sendPartialGroupOnExpiry 为 true 的组相同。如果它返回 false（在超时时），则根据 sendPartialGroupOnExpiry，该组被丢弃或释放。

expireGroupsUponTimeout默认为true，表示迟到的消息会组成一个新组，每条消息都会调用释放策略。如果您想丢弃迟到的消息，请将其设置为false；该组被保留（零消息），因此我们知道我们需要丢弃迟到者。

在这种情况下，对于迟到的消息不会调用释放策略，它们会立即被丢弃。

您可以通过检查组的大小轻松清理超时状态 - 如果与上次相同，则组已过期。

对于长期运行的系统，您可能希望最终使用消息组存储收割机使空组过期。