骡子聚合器-流聚合

Question

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

入站路由器

• 接收消息集合并将其拆分为多个较小的消息--每条较小的消息都被标记为对应于父消息
• 的相关id --这些消息流经各种
• ，这些消息到达一个入站聚合器，根据父消息的关联id和预期消息的数量收集消息。一旦接收到所有预期的消息，就会调用聚合函数并返回结果。

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

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

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

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

有更好的方法吗？

Answer 1

这似乎是一种合理的方法(我根本不是Mule专家)，而且我已经阅读了所有的Mule文档，而且我不认为有类似的东西(流支持仅限于几个连接器和变压器-它很简单，因为它只是通过一个InputStream)。在Mule流中只有一些东西，所以您可能需要其他修改的转换器(如果您使用它们)该流。您只需实现聚合器，即提供一个InputStream，并在它得到一些连续的消息序列后立即开始流。

但是，在您的描述中有一句话“.所有的消息都是在将结果转发到端点之前收到的”可能是令人不安的。从本质上讲，这违背了流的目的，除非您(可能在您的服务组件中)在转发(可能要小得多的)处理结果之前跟踪所有内容。