Java，通过多线程环境中的哈希将传入的工作均匀地划分

Question

我已经实现了一个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：

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

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

更新2：

具有相同任务的任务dispatchId可以多次出现，因此我们需要通过查找它们的线程dispatchId。

最重要的是，每个Worker线程必须按顺序处理其传入的任务。因此，上面的更新1中存在数据结构Queue。

更新3：
此外，某些线程可能很忙，而其他线程则空闲。因此，我们需要以某种方式将队列与线程解耦，但要保持FIFO顺序不变，dispatchId以便执行任务。

解决方案： 我已经实现了Ben
Manes的想法（他的回答如下），代码可以在这里找到。

Answer 1

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

提供这种抽象的非活动库是HawtDispatch。它与Java
6兼容。

一个非常简单的Java
8方法是使用CompletableFuture作为排队机制，使用ConcurrentHashMap进行注册，并使用执行程序（例如ForkJoinPool）进行计算。有关此想法的实现，请参阅EventDispatcher，其中注册是显式的。如果您的调度员更具动态性，那么您可能需要定期修剪地图。基本思想如下。

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）

上述想法的后盾将通过番石榴翻译成类似以下内容：

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();
  }
}