将 add() 和 take() 元素添加到 BlockingQueue 时死锁

Question

我正在玩 java.util.concurrent 包并使用它的 interface/classes 来了解它们是如何工作的。我创建了一个 BlockingQueue 实例（ArrayBlockingQueue 实现）。并创建了 Rannable 类型的 50 个消费者和 50 个生产者。然后 Executors.newFixedThreadPool(4) 创建了一个大小为 4 的线程池，并将我所有的 consumers/producers 提交给 threadPool (ExecutorService)。但最终打印过程我认为它是 deadlock-ing。谁能解释为什么线程安全队列会死锁！？下面是我的代码：

消费者：

public class ArrayBlockingQueueConsumer implements Runnable{

    BlockingQueue<Integer> blockingQueue;
    int consumerNumber = 0;

    public ArrayBlockingQueueConsumer(BlockingQueue<Integer> blockingQueue, int consumerNumber) {
        this.blockingQueue = blockingQueue;
        this.consumerNumber = consumerNumber;
    }

    public void run() {
        int i = 0;
        while(i<60) {
            System.out.printf("Consumer %d going take %d from blocking queue\n", consumerNumber, i);
            try {
                int x = blockingQueue.take();
                System.out.println("The number " + x + "is taken from the queue.");
            } catch (InterruptedException e) {
                System.out.printf("Consumer %d interrupted while adding %d to blocking queue\n", consumerNumber, i);
                e.printStackTrace();
            }
            i++;
        }
    }
}

制作人：

public class ArrayBlockingQueueProducer implements Runnable{

    BlockingQueue<Integer> blockingQueue;
    int producerNumber = 0;

    public ArrayBlockingQueueProducer(BlockingQueue<Integer> blockingQueue, int producerNumber) {
        this.blockingQueue = blockingQueue;
        this.producerNumber = producerNumber;
    }

    public void run() {
        int i = 0;
        while(i<60) {
            System.out.printf("Consumer %d going to add %d to blocking queue\n", producerNumber, i);
            blockingQueue.add(i);
            System.out.printf("Consumer %d added %d to blocking queue\n", producerNumber, i);
            i++;
        }
    }

}

执行器Class（main()方法class）：

public class BlockingQueueExecutor {

    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(50);

        ArrayBlockingQueueConsumer[] consumers = new ArrayBlockingQueueConsumer[200];
        ArrayBlockingQueueProducer[] producers = new ArrayBlockingQueueProducer[200];

        System.out.println("Hello hello :D");

        for (int i = 0; i < 200; i++) {
            consumers[i] = new ArrayBlockingQueueConsumer(blockingQueue, i+1);
            producers[i] = new ArrayBlockingQueueProducer(blockingQueue, i+1);
        }

        ExecutorService threadPool = Executors.newFixedThreadPool(4);

        for (int i = 0; i < 200; i++) {
            threadPool.execute(consumers[i]);
            threadPool.execute(producers[i]);
        }

        threadPool.shutdown();
    }
}

I 运行 consumers/producers 中的循环从 0 到 60 以便它们可以在分别找不到任何元素或发现队列已满时抛出异常，具有讽刺意味的是 none producers/consumers 引发任何异常。

Answer 1

I ran the loops in consumers/producers from 0 to 60 so that they can throw exception when they don't find any element or find queue full respectively, ironically none of the producers/consumers threw any exception.

当队列为空时，

take() 不会抛出异常。它一直等到元素可用。

E java.util.concurrent.BlockingQueue.take() throws InterruptedException

Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.

Answer 2

这里没有死锁。您的代码在 BlockingQueue.add() 调用中为我抛出一个 IllegalStateException。这导致队列没有像计划的那样满，最终导致 take() 到阻塞，如记录的那样。不是死锁。

死锁是指两个或多个线程无法继续执行的情况，因为它们都在等待彼此。

将 add() 和 take() 元素添加到 BlockingQueue 时死锁

Deadlock while add() and take() elements to BlockingQueue

java

multithreading

executorservice

blockingqueue

threadpool