阻塞队列和多线程消费者,如何知道何时停止

Question

我有一个单线程生成器,它创建一些任务对象,然后将其添加到ArrayBlockingQueue(具有固定大小).

我也开始了一个多线程的消费者.这是一个固定的线程池(Executors.newFixedThreadPool(threadCount);).然后我向这个threadPool提交了一些ConsumerWorker入口,每个ConsumerWorker都对上面提到的ArrayBlockingQueue实例进行了引用.

每个这样的工作人员都会take()在队列中执行并处理任务.

我的问题是,当没有更多的工作要做时,让工人知道的最佳方法是什么.换句话说,如何告诉Workers,生产者已经完成了对队列的添加,从这一点开始,每个工作人员在看到Queue为空时应该停止.

我现在得到的是一个设置,我的Producer初始化了一个回调,当他完成它的工作(向队列中添加东西)时会触发回调.我还保留了我创建并提交给ThreadPool的所有ConsumerWorkers的列表.当Producer Callback告诉我生产者已完成时,我可以告诉每个工人.此时,他们应该只是继续检查队列是否为空,当它变为空时它们应该停止,从而允许我优雅地关闭ExecutorService线程池.就是这样的

public class ConsumerWorker implements Runnable{

private BlockingQueue<Produced> inputQueue;
private volatile boolean isRunning = true;

public ConsumerWorker(BlockingQueue<Produced> inputQueue) {
    this.inputQueue = inputQueue;
}

@Override
public void run() {
    //worker loop keeps taking en element from the queue as long as the producer is still running or as 
    //long as the queue is not empty:
    while(isRunning || !inputQueue.isEmpty()) {
        System.out.println("Consumer "+Thread.currentThread().getName()+" START");
        try {
            Object queueElement = inputQueue.take();
            //process queueElement
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

//this is used to signal from the main thread that he producer has finished adding stuff to the queue
public void setRunning(boolean isRunning) {
    this.isRunning = isRunning;
}

}

这里的问题是我有一个明显的竞争条件,有时生产者将完成,发出信号,消费者工作者将在消耗队列中的所有内容之前停止.

我的问题是,同步这个的最佳方法是什么,以便一切正常？我应该同步整个部分来检查生产者是否正在运行加上如果队列是空的还加上一个块中的队列(在队列对象上)？我应该只是同步isRunningConsumerWorker实例上的布尔更新吗？还有其他建议吗？

更新,这里是我最终使用的工作实现:

public class ConsumerWorker implements Runnable{

private BlockingQueue<Produced> inputQueue;

private final static Produced POISON = new Produced(-1); 

public ConsumerWorker(BlockingQueue<Produced> inputQueue) {
    this.inputQueue = inputQueue;
}

@Override
public void run() {
    //worker loop keeps taking en element from the queue as long as the producer is still running or as 
    //long as the queue is not empty:
    while(true) {
        System.out.println("Consumer "+Thread.currentThread().getName()+" START");
        try {
            Produced queueElement = inputQueue.take();
            Thread.sleep(new Random().nextInt(100));
            if(queueElement==POISON) {
                break;
            }
            //process queueElement
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("Consumer "+Thread.currentThread().getName()+" END");
    }
}

//this is used to signal from the main thread that he producer has finished adding stuff to the queue
public void stopRunning() {
    try {
        inputQueue.put(POISON);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

}

这很大程度上受到了JohnVint在下面的回答的启发,只有一些小修改.

===由于@ vendhan的评论而更新.

谢谢你的观察.你是对的,这个问题的第一个代码片段(在其他问题中)是一个while(isRunning || !inputQueue.isEmpty())没有意义的代码片段.

在我实际的最终实现中,我做了一些更接近你更换"||"的建议.(或)用"&&"(和),从某种意义上说,每个工人(消费者)现在只检查他从列表中得到的元素是否是毒丸,如果是的话就停止了(理论上我们可以说工人有要运行并且队列不能为空).

Answer 1

你应该继续take()从队列中.你可以使用毒丸告诉工人停止.例如:

private final Object POISON_PILL = new Object();

@Override
public void run() {
    //worker loop keeps taking en element from the queue as long as the producer is still running or as 
    //long as the queue is not empty:
    while(isRunning) {
        System.out.println("Consumer "+Thread.currentThread().getName()+" START");
        try {
            Object queueElement = inputQueue.take();
            if(queueElement == POISON_PILL) {
                 inputQueue.add(POISON_PILL);//notify other threads to stop
                 return;
            }
            //process queueElement
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

//this is used to signal from the main thread that he producer has finished adding stuff to the queue
public void finish() {
    //you can also clear here if you wanted
    isRunning = false;
    inputQueue.add(POISON_PILL);
}

Answer 2

我会给工人们发一个特殊的工作包来表示他们应该关闭:

public class ConsumerWorker implements Runnable{

private static final Produced DONE = new Produced();

private BlockingQueue<Produced> inputQueue;

public ConsumerWorker(BlockingQueue<Produced> inputQueue) {
    this.inputQueue = inputQueue;
}

@Override
public void run() {
    for (;;) {
        try {
            Produced item = inputQueue.take();
            if (item == DONE) {
                inputQueue.add(item); // keep in the queue so all workers stop
                break;
            }
            // process `item`
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

}

要停止工作人员,只需添加ConsumerWorker.DONE到队列中即可.