聚合器，它根据相关性释放部分组，但保留其余消息

Question

我想在聚合器上设置相关策略，以便它使用传入文件（作为消息）名称之外的日期来关联文件，以便所有具有今天日期的文件都属于同一组。现在，由于我可能有多天的数据，我可能已经聚合了2天的文件。我想把发布策略建立在一个已完成的文件（消息）的基础上，该文件包括文件名中的日期，因此基本上每天都会有一堆文件和一个已完成的文件。摄取完成的文件应该从聚合器中释放当天的文件，但仍然保留其他日期的文件，直到摄取当天的完成文件为止。

因此，在这个场景中，关联显然很简单，但我不确定的是如何根据关联键发布组中的所有消息，而只是一些特定消息。文档谈到了messagereaper，但这涉及到messagestore的内容，我想在内存中完成所有这些。

让我用一个例子来说明

我将这些文件放在一个目录中，该目录由文件入站通道适配器进行im轮询

文件-1-2014.04.27。dat公司

文件-2-2014.04.27。dat公司

文件-3-2014.04.27。dat公司

done-2014.04.27.dat

file-1-2014.04.28.dat

file-2-2014.04.28.dat

done-2014.04.28.dat

由于这些文件正在轮询中，我在流中有一个聚合器，所有传入的文件都在其中聚合。为了关联，我想我可以提取日期并将其放在correlation_id头中，以便前3个文件被认为属于一个组，然后第2个文件属于第二组...现在，一旦我当时使用了done-2014.04.27.dat文件，我想释放前3个文件以在流中进一步处理，但请坚持

file-1-2014.04.28.dat

file-2-2014.04.28.dat

直到我收到

完成日期：2014年4月28日。dat公司

然后释放这2个文件。

任何帮助都将不胜感激。谢谢

Answer 1

我不知道你说的“相关性很简单”是什么意思，但接着又说，你只想释放团队的一部分。如果他们有不同的日期，那么他们将在不同的组中，因此不需要释放组中的一部分，只需在午夜后（或第二天的任何时间）运行收割器来释放整个组。你为什么需要一个“完成”的消息，这一点都不清楚。

默认情况下，聚合器使用内存中的消息存储（SimpleMessageStore）。

编辑：

只需将完成文件放在同一个组中，并让您的发布策略检测完成文件的存在。您可以使用表达式，但如果组可以很大，则实现ReleaseStrategy并遍历MessageGroup.getMessages（）寻找完成文件会更有效。

下一步取决于聚合器的下游。如果使用拆分器将它们拆分回单独的文件，只需添加一个过滤器即可删除完成的文件。如果直接处理文件集合，请忽略已完成的文件，或添加转换器将其从集合中删除。

关于收割者；假设文件实时到达，我只是建议，如果你每天运行一次收割者（比如01:00），组超时时间为30分钟，那么收割者将释放昨天的文件（不需要完成的文件）。

编辑：

请参阅下面我对您的“答案”的评论-您在filesLogger上有2个订阅者。