Mule 聚合器 - 流式聚合

Question

Mule 2.0 框架中使用的集合聚合器有点像这样：

• 入站路由器获取消息集合并将其拆分为多个较小的消息 - 每个较小的消息都带有与父消息对应的相关 ID 标记

• 这些消息流经各种服务

• 最终这些消息到达一个入站聚合器，该聚合器根据父消息的相关 ID 和预期消息的数量收集消息。一旦收到所有预期的消息，就会调用聚合函数并返回结果。

现在，当组中的消息数量相当少时，这可以正常工作。然而，一旦一个组中的消息数量变得巨大~100k，那么就会有很多内存被占用来等待后面的消息到达的消息组。如果同时聚合多个组，情况会更糟。

解决此问题的一种方法是实现流式聚合器。在我的用例中，我基本上是基于一个键来总结各种消息，这可以在不必同时查看组中的所有消息的情况下完成。我只想知道在将结果转发到端点之前已收到所有消息。

这听起来像是解决问题的合理方法吗？

这是否已经在 Mule 的某个地方实现了？

有没有更好的方法来做到这一点？

Answer 1

这似乎是一种合理的方法（无论如何我都不是 Mule 专家），而且我已经阅读了所有 Mule 文档并且认为那里没有类似的东西（流媒体支持仅限于一些连接器和转换器——它非常简单，它只是绕过一个 InputStream）。 Mule 流中只有一些东西，因此您可能需要其他修改过的转换器（如果您使用它们）该流。您只需实现提供 InputStream 的聚合器，并在它收到一些连续的消息序列后立即开始流式传输。

但是，您的描述中的一句话“......在将结果转发到端点之前已收到所有消息”可能会令人不安。从本质上讲，这违背了流式传输的目的，除非您的意思是您（大概在您的服务组件中）将跟踪您在转发（大概更小）处理结果之前获得的所有内容。