我正在实施标准的生产者－消费者计划，以使生产者在生产200种产品后停止运转。为了表明这一点，生产者将-1放入BlockingQueue变量，否则该变量始终包含正整数。
我的消费者实现如下：
public class Consumer implements Runnable{
private BlockingQueue<Integer> queue;
private AtomicBoolean isProducerClosed = new AtomicBoolean(false);
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
Thread.sleep(50);
while(!isProducerClosed.get()) {
try {
Integer value = queue.take();
if ((value.intValue() == -1)){
System.out.println(Thread.currentThread().getName() + " Encountered -1. Signal to shutdown consumers.");
//isProducerClosed.set(true);
isProducerClosed.compareAndSet(false, true);
break;
}
System.out.println(Thread.currentThread().getName() + " consuming : " + value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
替代消费者逻辑（仍然出现相同问题）：
@Override
public void run() {
while(true) {
try {
Thread.sleep(50);
Integer value = null;
synchronized (this) {
if(!isProducerClosed.get()) {
value = queue.take();
if ((value.intValue() == -1)) {
System.out.println(Thread.currentThread().getName() + " Encountered -1. Signal to shutdown consumers.");
isProducerClosed.set(true);
isProducerClosed = isProducerClosed;
System.out.println("Current value of isProducerClosed : " + isProducerClosed.get());
}
}
}
if (!isProducerClosed.get()) {
System.out.println(Thread.currentThread().getName() + " consuming : " + value);
} else {
break;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
执行此操作时，我的使用者线程被卡在
queue.take()上，就像他们在等待队列中的产品可用一样。一种可能是：所有使用者线程同时检查条件isProducerClosed.get()，进入while循环，并访问queue.take()方法。这是正确的假设吗？如果是，是否有任何方法可以在不使用低级synchronized关键字的情况下实现此目的？我用
volatile boolean变量尝试了同样的事情，结果是完全一样的。只有在将该变量设为静态之后，我才能看到所有使用者在队列中遇到-1后都被终止（因为var现在是类拥有的）。我的通话代码：
public class ProducerConsumer {
private BlockingQueue<Integer> sharedQueue = new ArrayBlockingQueue<Integer>(10);
public void executeProducerConsumer(int producerCount, int consumerCount){
ExecutorService executorService = Executors.newFixedThreadPool(3, Executors.defaultThreadFactory());
for(int i = 0; i < producerCount; i++) {
executorService.submit(new Producer(sharedQueue)); //producer
}
for(int i = 0; i < consumerCount; i++) {
executorService.submit(new Consumer(sharedQueue)); //i-th consumer.
}
//initiates clossure of threads after completion, in async manner.
executorService.shutdown();
//wait till all threads are done.
try {
executorService.awaitTermination(1, TimeUnit.HOURS);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("The End.");
}
}
生产者代码：
public class Producer implements Runnable {
private BlockingQueue<Integer> queue;
private volatile int maxNumberOfItemsToProduce = 10;
public Producer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
Random random = new Random();
while(true){
if(maxNumberOfItemsToProduce == 0) {
try {
queue.put(-1);
System.out.println("Terminating Producer after producing max number of products.");
} catch (InterruptedException e) {
e.printStackTrace();
}
break;
}
try {
queue.put(random.nextInt(300));
} catch (InterruptedException e) {
e.printStackTrace();
}
maxNumberOfItemsToProduce--;
}
}
}
控制台的屏幕截图：
最佳答案
一种可能是：所有使用者线程同时检查条件isProducerClosed.get（），进入while循环，并访问queue.take（）方法。这是正确的假设吗？如果是，是否有任何方法可以在不使用低级同步关键字的情况下实现此目的？我用volatile布尔变量尝试了同样的事情，结果是完全一样的。只有在将该变量设为静态之后，我才能看到所有使用者在队列中遇到-1后都被终止（因为var现在是类拥有的）。
是的，这个假设是正确的。
第一个问题是：消费者之间没有共享isProducerClosed。它必须在消费者之间共享，这样，如果一个消费者设定了它的价值，其他消费者也可以看到该价值。将其设置为static使其共享，因此此后情况有所改善
第二个问题：即使共享了isProducerClosed，您也可能会遇到多个使用者在空队列上执行queue.take()的情况（一个线程可能获取最后一个值，而另一个线程则在第一个执行之前执行take()将isProducerClosed设置为true）。您将需要对此进行同步（例如，使用双重检查）
示例代码（仍然包含使用者以外的部分中的错误/竞赛）-
public class TestClass {
private BlockingQueue<Integer> sharedQueue = new ArrayBlockingQueue<Integer>(10);
public static void main(String[] args) {
TestClass t = new TestClass();
t.executeProducerConsumer(3, 3);
}
public void executeProducerConsumer(int producerCount, int consumerCount) {
ExecutorService executorService = Executors.newFixedThreadPool(producerCount + consumerCount, Executors.defaultThreadFactory());
for (int i = 0; i < producerCount; i++) {
executorService.submit(new Producer(sharedQueue)); //producer
}
for (int i = 0; i < consumerCount; i++) {
executorService.submit(new Consumer(sharedQueue)); //i-th consumer.
}
//initiates clossure of threads after completion, in async manner.
executorService.shutdown();
//wait till all threads are done.
try {
executorService.awaitTermination(1, TimeUnit.HOURS);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("The End.");
}
}
class Consumer implements Runnable {
private BlockingQueue<Integer> queue;
private static volatile boolean isProducerClosed = false; // make this static so that it is shared across consumers
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
Thread.sleep(50);
Integer value;
while (!isProducerClosed) {
try {
synchronized (queue) { //synchronize so that only one thread can talk to the queue at a time
if (!isProducerClosed) { //double check
value = queue.take(); // we can now safely take an item
if ((value.intValue() == -1)) {
isProducerClosed = true;
System.out.println(Thread.currentThread().getName() + " Encountered -1. Signal to shutdown consumers.");
break;
}
} else {
System.out.println(Thread.currentThread().getName() + " Last item was taken by some other consumer. Exiting!");
break;
}
}
consumeValue(value); //Consume the value outside the lock
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private void consumeValue(Integer value) {
System.out.println(Thread.currentThread().getName() + " Consuming value :" + value);
}
}
class Producer implements Runnable {
private BlockingQueue<Integer> queue;
private static volatile int maxNumberOfItemsToProduce = 10;
public Producer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
Random random = new Random();
while (true) {
if (maxNumberOfItemsToProduce == 0) {
try {
queue.put(-1);
System.out.println("Terminating Producer after producing max number of products.");
} catch (InterruptedException e) {
e.printStackTrace();
}
break;
}
try {
queue.put(random.nextInt(300));
} catch (InterruptedException e) {
e.printStackTrace();
}
maxNumberOfItemsToProduce--;
}
}
}