如何处理可能创建新工作项的多个工作线程

Question

我有一个包含工作项的队列，我希望多个线程在这些项上并行工作。 处理工作项时，可能会产生新的工作项。我遇到的问题是我找不到如何确定是否完成的解决方案。工人看起来像这样：

public class Worker implements Runnable {
  public void run() {
    while (true) {
      WorkItem item = queue.nextItem();
      if (item != null) {
        processItem(item);
      }
      else {
        // the queue is empty, but there may still be other workers
        // processing items which may result in new work items
        // how to determine if the work is completely done?
      }
    }
  }
}

这实际上似乎是一个非常简单的问题，但我不知所措。实现它的最佳方式是什么？

谢谢

说明： 一旦没有一个工作线程正在处理某个项目，工作线程就必须终止，但只要其中至少一个仍在工作，它们就必须等待，因为这可能会导致新的工作项目。

Answer 1

如何使用 ExecutorService 让您等待所有任务完成：ExecutorService, how to wait for all tasks to finish

Answer 2

我建议等待/通知电话。在其他情况下，您的工作线程将等待一个对象，直到队列通知还有更多工作要做。当工作人员创建一个新项目时，它将其添加到队列中，并且队列对工作人员正在等待的对象调用通知。其中一个将醒来以消耗新项目。

Object 类的 wait、notify 和 notifyAll 方法支持从一个线程到另一个线程的有效控制转移。与消耗计算量的简单“旋转”（反复锁定和解锁对象以查看某些内部状态是否已更改）不同，线程可以使用等待挂起自己，直到另一个线程使用通知唤醒它。这尤其适用于线程具有生产者-消费者关系（为共同目标积极合作）而不是互斥关系（试图在共享公共资源时避免冲突）的情况。

Source: Threads and Locks

Answer 3

我会看比等待/通知更高级别的东西。很难做到正确并避免死锁。你看过java.util.concurrent.CompletionService<V>吗？你可以有一个更简单的管理器线程来轮询服务和take()s 结果，它可能包含也可能不包含新的工作项。

Answer 4

使用包含要处理的项目的 BlockingQueue 以及跟踪当前正在处理的所有元素的同步集：

BlockingQueue<WorkItem> bQueue;
Set<WorkItem> beingProcessed = new Collections.synchronizedSet(new HashMap<WorkItem>());
bQueue.put(workItem);
...

// the following runs over many threads in parallel
while (!(bQueue.isEmpty() && beingProcessed.isEmpty())) {
  WorkItem currentItem = bQueue.poll(50L, TimeUnit.MILLISECONDS); // null for empty queue
  if (currentItem != null) {
    beingProcessed.add(currentItem);   
    processItem(currentItem); // possibly bQueue.add(newItem) is called from processItem
    beingProcessed.remove(currentItem);
  }
}

编辑：正如@Hovercraft Full Of Eels 建议的那样，ExecutorService 可能是您真正应该使用的。您可以随时添加新任务。您可以使用executorService.awaitTermination(time, timeUnits) 定期半忙等待所有任务的终止，然后杀死所有线程。

Answer 5

这是解决您的问题的队列的开始。基本上，您需要跟踪新工作和正在进行的工作。

public class WorkQueue<T> {
    private final List<T> _newWork = new LinkedList<T>();
    private int _inProcessWork;

    public synchronized void addWork(T work) {
        _newWork.add(work);
        notifyAll();
    }

    public synchronized T startWork() throws InterruptedException {
        while(_newWork.isEmpty() && (_inProcessWork > 0)) {
            wait();
            if(!_newWork.isEmpty()) {
                _inProcessWork++;
                return _newWork.remove(0);
            }
        }
        // everything is done
        return null;
    }

    public synchronized void finishWork() {
        _inProcessWork--;
        if((_inProcessWork == 0) && _newWork.isEmpty()) {
            notifyAll();
        }
    }
}

您的工人大致如下：

public class Worker {
  private final WorkQueue<T> _queue;

  public void run() {
    T work = null;
    while((work = _queue.startWork()) != null) {
      try {
        // do work here...
      } finally {
        _queue.finishWork();
      }
    }
  }
}

一个技巧是您需要在启动任何工作人员之前添加第一个工作项（否则他们将立即退出）。