将给定 ID 的任务绑定到同一线程的线程池

Question

是否有线程池（Java 中）的实现可以确保同一逻辑 ID 的所有任务都在同一线程上执行？

我所遵循的逻辑是，如果给定逻辑 ID 的特定线程上已经执行了一个任务，则具有相同 ID 的新任务将被安排在同一线程上。如果没有线程执行相同 ID 的任务，则可以使用任何线程。

这将允许不相关 ID 的任务并行执行，但相同 ID 的任务按照提交的顺序串行执行。

如果没有，是否有关于如何扩展 ThreadPoolExecutor 来获得此行为的建议（如果可能的话）？

更新

经过更长时间的思考，我实际上并不要求相同逻辑 ID 的任务在同一线程上执行，只是它们不会同时执行。

一个例子是为客户处理订单的系统，可以同时处理多个订单，但不能为同一客户处理（同一客户的所有订单都必须按顺序处理）。

我目前采用的方法是使用标准 ThreadPoolExecutor，带有自定义的 BlockingQueue，并使用自定义包装器包装 Runnable。 Runnable 包装器逻辑是：

1. 以原子方式尝试将 ID 添加到并发“运行”集 (ConcurrentHashMap)，以查看当前是否正在运行相同 ID 的任务
   • 如果添加失败，则将任务推回到队列前面并立即返回
   • 如果成功，继续
2. 运行任务
3. 从“运行”集合中删除任务关联的ID

队列的poll()< /code> 方法则仅返回 ID 当前不在“正在运行”集中的任务。

这样做的问题是，我确信会有很多我没有考虑到的极端情况，因此需要大量测试。

Answer 1

创建一组执行程序服务，每个执行程序服务运行一个线程，并通过项目 ID 的哈希码将队列条目分配给它们。该数组可以是任意大小，具体取决于您最多要使用多少个线程。

这将限制我们可以从执行程序服务中使用，但仍然允许使用其功能在不再需要时关闭唯一的线程（使用allowCoreThreadTimeOut(true)）并根据需要重新启动它。此外，所有排队的东西都可以工作而无需重写。

Answer 2

最简单的想法可能是这样的：

拥有一个固定的 BlockingQueue 映射。使用哈希机制根据任务 ID 选择队列。哈希算法应该为相同的 id 选择相同的队列。为每个队列启动一个线程。每个线程都会从它自己的专用队列中选择一个任务并执行它。

ps 适当的解决方案很大程度上取决于您分配给线程的工作类型

UPDATE

好吧，这个疯狂的想法怎么样，请耐心等待:)

比如说，我们有一个 ConcurrentHashMap 包含引用 id -> OrderQueue

ID1->Q1, ID2->Q2, ID3->Q3, ...

意味着现在每个 id 都与其自己的队列相关联。 OrderQueue 是一个自定义阻塞队列，带有附加布尔标志 - isAssociatedWithWorkingThread。

还有一个常规的 BlockingQueue，我们现在将其称为 amortizationQueue，稍后您将看到它的使用。

接下来，我们有 N 个工作线程。每个工作线程都有自己的工作队列，它是一个包含与该线程关联的 ID 的 BlockingQueue。

当新的 id 出现时，我们执行以下操作：

create a new OrderQueue(isAssociatedWithWorkingThread=false)
put the task to the queue
put id->OrderQueue to the map
put this OrderQueue to amortizationQueue

当现有 id 的更新出现时，我们执行以下操作：

pick OrderQueue from the map
put the task to the queue
if isAssociatedWithWorkingThread == false
    put this OrderQueue to amortizationQueue

每个工作线程执行以下操作：

take next id from the working queue
take the OrderQueue associated with this id from the map
take all tasks from this queue
execute them
mark isAssociatedWithWorkingThread=false for this OrderQueue
put this OrderQueue to amortizationQueue

非常简单。现在到了有趣的部分 - 工作窃取:)

如果在某个时间点某个工作线程发现自己的工作队列为空，那么它会执行以下操作：

go to the pool of all working threads
pick one (say, one with the longest working queue)
steal id from *the tail* of that thread's working queue
put this id to it's own working queue
continue with regular execution

并且还有 +1 个额外线程提供摊销工作：

while (true)
    take next OrderQueue from amortizationQueue
    if queue is not empty and isAssociatedWithWorkingThread == false
         set isAssociatedWithWorkingThread=true
         pick any working thread and add the id to it's working queue

将不得不花费更多时间考虑是否可以使用 AtomicBoolean 作为 isAssociatedWithWorkingThread 标志，或者需要使其阻塞操作来检查/更改此标志。

Answer 3

我最近不得不处理类似的情况。

我最终得到了与你类似的设计。唯一的区别是“当前”是一个映射而不是一个集合：从 ID 到 Runnables 队列的映射。当任务的可运行对象的包装器发现其 ID 存在于映射中时，它会将任务的可运行对象添加到 ID 的队列中并立即返回。否则，ID 将被添加到具有空队列的映射中并执行任务。

任务完成后，包装器会再次检查 ID 的队列。如果队列不为空，则选择可运行的。否则它就会从地图上删除，我们就完成了。

我将把关闭和取消作为练习留给读者:)

Answer 4

我们的方法与原始问题的更新中的方法类似。我们有一个可运行的包装类，其中包含一个队列（LinkedTransferQueue），我们将其称为 RunnableQueue。 Runnable 队列具有以下基本 API：

public class RunnableQueue implements Runnable
{
  public RunnableQueue(String name, Executor executor);
  public void run();

  public void execute(Runnable runnable);
}

当用户通过执行调用提交第一个 Runnable 时，RunnableQueue 将自身排队到执行器上。后续对执行的调用会在 RunnableQueue 内的队列中排队。当可运行队列被 ThreadPool 执行（通过其 run 方法）时，它开始通过逐个串行执行可运行队列来“耗尽”内部队列。如果在执行时对 RunnableQueue 调用execute，则新的可运行对象将简单地附加到内部队列中。一旦队列被清空，可运行队列的 run 方法就会完成并“离开”执行器池。重复冲洗。

我们还有其他优化，例如在 RunnableQueue 将自身重新发布到执行程序池之前只让一定数量的可运行程序运行（例如四个）。

里面唯一真正棘手的一点（而且并不那么难）是在它被发布到执行器或不发布到执行器时进行同步，这样它就不会重新发布，或者错过它应该发布的时间。

总的来说，我们发现这个方法运行得很好。我们的“ID”（语义上下文）是可运行队列。我们的需求（即插件）引用了 RunnableQueue 而不是执行器池，因此它被迫通过 RunnableQueue 专门工作。这不仅保证所有访问都是连续的（线程限制），而且让 RunnableQueue “调节”插件的作业加载。此外，它不需要集中管理结构或其他争论点。

Answer 5

我必须实现一个类似的解决方案，并且通过 h22 创建一组执行器服务的建议对我来说似乎是最好的方法，但需要注意的是，我将采用 ID 的模数 % （或者是原始数据） ID（假设它是 long/int 或哈希码）相对于某些所需的最大大小，并使用该结果作为新 ID，这样我就可以在最终获得太多执行程序服务对象同时仍然获得大量执行程序服务对象之间取得平衡中的并发数 加工。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorServiceRouter {

    private List<ExecutorService> services;
    private int size;

    public ExecutorServiceRouter(int size) {
        services = new ArrayList<ExecutorService>(size);
        this.size = size;
        for (int i = 0; i < size; i++) {
            services.add(Executors.newSingleThreadExecutor());
        }
    }

    public void route(long id, Runnable r) {
        services.get((int) (id % size)).execute(r);
    }

    public void shutdown() {
        for (ExecutorService service : services) {
            service.shutdown();
        }
    }

}

Answer 6

扩展 ThreadPoolExecutor 会非常困难。我建议你采用生产者-消费者系统。这就是我的建议。

1. 您可以创建典型的生产者消费者系统。查看此问题中提到的代码。
2. 现在，每个系统都会有一个队列和一个单一消费者线程，它将串行处理队列中的任务。
3. 现在，创建一个此类单独系统的池。
4. 当您提交相关 ID 的任务时，请查看是否已经有一个为该相关 ID 标记的系统当前正在处理该任务，如果是，则提交任务，
5. 如果它没有处理任何任务，则用这个新的相关 ID 标记该系统ID并提交任务。
6. 这样，单个系统将只满足一个逻辑相关的 ID。

在这里，我假设相关 ID 是单个 ID 的逻辑集合，并且将为相关 ID 而不是单个 ID 创建生产者消费者系统。