并行执行和聚合器锁定

Question

我使用spring集成来处理文件的一些目录，每个文件都通过一个“流”。我希望以这样一种方式设置文件的整体处理，即文件轮询器监视一个目录（或多个目录）中的新文件。一旦轮询器拾取了一个新文件，就应该将其传递给流中的后续组件，在该组件中处理该新文件，而不保持轮询过程。处理的第二个方面是，所有新文件都要经过几个步骤，然后由聚合器根据文件数（标准在不同目录中变化）进行聚合。一旦积累了足够的文件，就可以从聚合中释放它们，然后在聚合器之后以一些耗时的步骤进行处理。整个过程如下所示

file-A拾取的
file-A从轮询器传递到step1
file-A从step1传递到聚合器
file-B拾取的
file-B从轮询器传递到step1
file-B从step1传递到聚合器
file-C拾取的
file-C从轮询器传递到step1
file-C从step1传递到聚合器

文件A、B和C从聚合器
中释放文件A、B和C由final-Step
处理

所以总体来说有两个要求

1. 在单独的线程中处理每个文件，以最大限度地增加当前正在处理的文件数
2. 从聚合器发布的文件属于一个关联id，我们只希望在最后一步处理一组使用相同关联id的消息

我如何尝试满足这两个要求是为了#1，我只是在文件轮询器之后使用一个队列，在该队列中丢弃新文件，然后步骤a从队列中拾取文件。这分离了轮询过程，其想法是在步骤中使用线程执行器—服务激活器在单个线程中处理每个文件—第二个需求是通过在聚合器所在的线程中，在聚合器之后执行最后一个步骤来自动处理的。由于聚合器在释放另一个组以获得相同的相关id时会基于相关id设置锁，因此它只需等待相同组的前一个实例被处理

我遇到的问题是#1没有完成，因为服务激活器在尝试为第二个文件创建另一个线程之前，一直等到线程完成结束。这并没有什么帮助，因为在service activator上有一个线程执行器并没有什么用处。似乎只有在完成第一个线程后才能创建第二个therad。因此，为了解决这个问题，我将排队通道替换为调度程序通道，并在调度程序通道上分配执行器。现在，每个文件都在一个单独的线程中处理，同时处理多个文件<现在是第二部分，由于聚合器之后的组件非常耗时，我想将该html" target="_blank">进程与第一部分断开连接，所以我在聚合器之后放置了一个排队通道，但现在使用这种方法，我以前在聚合器中获得的锁定行为消失了，因为在最后耗时的步骤之前，从聚合器释放消息的线程在排队通道中死亡/完成。

对整个过程的任何想法。我如何在并行运行的同时完成我的两个需求。

谢谢

Answer 1

根本不清楚你的问题是什么。是的，为了使下游流在锁下运行，它必须在最终的“释放”线程（处理完成组的最后一条入站消息的线程）上运行，您不能在聚合器的下游使用队列或执行器通道。

但是，这对来自执行器通道的线程没有影响；其他组（具有不同相关性）将进行处理。但是，如果您对“next”组使用相同的关联id，则其线程将被阻塞。

如果您想在第一个组正在下游处理时组装下一个组（使用相同的correlationid），则必须使用其他机制来强制下游单线程，例如带有单线程执行器的执行器通道，或使用另一个锁注册表。