Java，在多线程环境中通过哈希统一划分传入的工作

我已经实现了一个java代码来执行传入的任务（如Runnable) 具有基于 hashCode 模块的 n 个线程nThreads。理想情况下，工作应该在这些线程之间均匀地分布。 具体来说，我们有一个dispatchId作为每个任务的字符串。

这是这个java代码片段：

int nThreads = Runtime.getRuntime().availableProcessors(); // Number of threads
Worker[] workers = new Worker[nThreads]; // Those threads, Worker is just a thread class that can run incoming tasks
...
Worker getWorker(String dispatchId) { // Get a thread for this Task
    return workers[(dispatchId.hashCode() & Integer.MAX_VALUE) % nThreads];
}

重要的：在大多数情况下，dispatchId 是：

String dispatchId = 'SomePrefix' + counter.next()

但是，我担心 nThreads 的模除不是一个好的选择，因为 nThreads 应该是素数，以便更均匀地分布 dispatId 键。

对于如何更好地开展工作还有其他选择吗？

更新1：

每个Worker都有一个队列：Queue<RunnableWrapper> tasks = new ConcurrentLinkedQueue();

工作人员从中获取任务并执行它们。可以从其他线程将任务添加到此队列。

更新2：

任务相同dispatchId可以多次出现，因此我们需要通过以下方式找到他们的线程dispatchId.

最重要的是，每个工作线程必须顺序处理其传入的任务。因此，上面的更新1中就有了数据结构Queue。

更新3：此外，某些线程可能很忙，而另一些线​​程则空闲。因此，我们需要以某种方式将队列与线程解耦，但保持相同的 FIFO 顺序dispatchId用于任务执行。

解决方案：我已经实现了 Ben Manes 的想法（他的答案如下），代码可以找到here https://github.com/vibneiro/dispatching.

听起来您需要每个调度 ID 进行 FIFO 排序，因此理想的情况是将调度队列作为抽象。这可以解释您对散列的担忧，因为它不提供均匀分布，因为某些调度队列可能比其他队列更活跃，并且在工作人员之间不公平地平衡。通过将队列与工作线程分离，您可以保留 FIFO 语义并均匀地分散工作。

提供这种抽象的非活动库是轴调度 https://github.com/fusesource/hawtdispatch。它与 Java 6 兼容。

一个非常简单的 Java 8 方法是使用完整的未来 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html作为排队机制，并发哈希映射 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentHashMap.html注册和执行人（例如分叉连接池 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ForkJoinPool.html）用于计算。看事件调度器 https://github.com/ben-manes/caffeine/blob/master/jcache/src/main/java/com/github/benmanes/caffeine/jcache/event/EventDispatcher.java为了实现这个想法，注册是明确的。如果您的调度员更加动态，那么您可能需要定期修剪地图。基本思路如下。

ConcurrentMap<String, CompletableFuture<Void>> dispatchQueues = ...

public CompletableFuture<Void> dispatch(String queueName, Runnable task) {
  return dispatchQueues.compute(queueName, (k, queue) -> {
    return (queue == null)
        ? CompletableFuture.runAsync(task)
        : queue.thenRunAsync(task);
  });
}

更新（JDK7）

上述想法的反向移植将用 Guava 翻译成类似的东西：

ListeningExecutorService executor = ...
Striped<Lock> locks = Striped.lock(256);
ConcurrentMap<String, ListenableFuture<?>> dispatchQueues = ...

public ListenableFuture<?> dispatch(String queueName, final Runnable task) {
  Lock lock = locks.get(queueName);
  lock.lock();
  try {
    ListenableFuture<?> future = dispatchQueues.get(queueName);
    if (future == null) {
      future = executor.submit(task);
    } else {
      final SettableFuture<Void> next = SettableFuture.create();
      future.addListener(new Runnable() {
        try {
          task.run();
        } finally {
          next.set(null);
        }
      }, executor);
      future = next;
    }
    dispatchQueues.put(queueName, future);
  } finally {
    lock.unlock();
  }
}