具有停止条件的 Java 生产者-消费者

Question

我有 N 个工作人员共享一个要计算的元素队列。在每次迭代中，每个工作人员从队列中删除一个元素，并可以生成更多要计算的元素，这些元素将被放入同一个队列中。基本上，每个生产者也是消费者。当队列中没有元素并且所有 worker 都已完成当前元素的计算时，计算结束(因此无法生成更多要计算的元素)。我想避免调度员/协调员，所以工作人员应该协调。允许工作人员确定停止条件是否有效并因此代表其他人停止计算的最佳模式是什么？

例如，如果所有线程都只做这个循环，当所有元素都计算完时，将导致所有线程永远阻塞:

while (true) {
    element = queue.poll();
    newElements[] = compute(element);
    if (newElements.length > 0) {
        queue.addAll(newElements);
    }
}

Answer 1

维护 Activity 线程的数量。

public class ThreadCounter {
    public static final AtomicInteger threadCounter = new AtomicInteger(N);
    public static final AtomicInteger queueCounter = new AtomicInteger(0);
    public static final Object poisonPill = new Object();
    public static volatile boolean cancel = false; // or use a final AomticBoolean instead
}

您的线程的轮询循环应如下所示(我假设您使用的是 BlockingQueue)

while(!ThreadCounter.cancel) {
    int threadCount = ThreadCounter.threadCounter.decrementAndGet(); // decrement before blocking
    if(threadCount == 0 && ThreadCounter.queueCounter.get() == 0) {
        ThreadCounter.cancel = true;
        queue.offer(ThreadCounter.poisonPill);
    } else {
        Object obj = queue.take();
        ThreadCounter.threadCounter.incrementAndGet(); // increment when the thread is no longer blocking
        ThreadCounter.queueCounter.decrementAndGet();
        if(obj == ThreadCounter.poisonPill) {
            queue.offer(obj); // send the poison pill back through the queue so the other threads can read it
            continue;
        }
    }
}

如果一个线程将要在 BlockingQueue 上阻塞，那么它会递减计数器；如果所有线程都已经在等待队列(意味着 counter == 0)，那么最后一个线程将 cancel 设置为 true，然后通过队列发送毒丸到唤醒其他线程；每个线程看到毒丸，通过队列将其发回以唤醒其余线程，然后在看到 cancel 设置为 true 时退出循环。

编辑:我通过添加一个 queueCounter 来消除数据竞争，它维护队列中对象数量的计数(显然您还需要在将对象添加到队列的任何位置添加一个 queueCounter.incrementAndGet() 调用)。其工作方式如下:如果 threadCount == 0，但 queueCount != 0，则这意味着线程刚刚从队列中删除了一个项目但尚未调用threadCount.getAndIncrement，因此取消变量未设置为 true。 threadCount.getAndIncrement 调用先于 queueCount.getAndDecrement 调用，这一点很重要，否则您仍然会有数据竞争。您调用 queueCount.getAndIncrement 的顺序无关紧要，因为您不会将它与 threadCount.getAndDecrement 的调用交织在一起(后者将在循环结束时，前者将在循环开始时被调用。

请注意，您不能只使用 queueCount 来确定何时结束进程，因为线程可能仍然处于 Activity 状态，但尚未将任何数据放入队列 - 换句话说，queueCount 将为零，但一旦线程完成其当前迭​​代就会为非零。

您可以让取消线程通过队列发送 (N-1) poisonPills，而不是通过队列重复发送 poisonPill。如果您通过不同的队列使用此方法，请务必小心，因为某些队列(例如亚马逊的简单队列服务)可能会返回多个与其 take 方法等效的项目，在这种情况下，您需要重复发送 poisonPill 以确保一切都关闭。

此外，您可以使用 while(true) 循环，而不是使用 while(!cancel) 循环，并在循环检测到 poisonPill 时中断