5

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

例如，如果所有线程都只是执行这个循环，那么当所有元素都被计算出来时，这将导致所有线程永远被阻塞：

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

1 回答 1

6

保持活动线程的计数。

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()向队列中添加对象的任何位置添加一个调用）。其工作原理如下： if threadCount == 0, but queueCount != 0，则这意味着线程刚刚从队列中删除了一项但尚未调用threadCount.getAndIncrement，因此取消变量未设置为 true。threadCount.getAndIncrement调用之前调用很重要queueCount.getAndDecrement，否则您仍然会有数据竞争。您调用的顺序无关紧要，queueCount.getAndIncrement因为您不会将其与调用交错threadCount.getAndDecrement（后者将在循环结束时调用，前者将在循环开始时调用）。

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

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

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

于 2013-05-16T16:36:09.707 回答