Java - BlockingQueue 卡住多线程应用程序

Question

我正在创建一个包含两个线程的应用程序:其中一个将值写入 LinkedBlockingQueue，另一个正在读取。我正在使用 ScheduledExecutorService 在几秒钟内运行此操作。 问题是我的应用程序在 BlockingQueue 的方法上卡住，我不明白为什么。

这是一个公共(public)资源:

class Res{
    AtomicInteger atomicInteger = new AtomicInteger(0);
    BlockingQueue<String> q = new LinkedBlockingQueue<>();
}

这是阅读器

Semaphore semaphore = new Semaphore(1); /this is for reader does not take two places in thread pool
Runnable reader = ()->{
    try {
        semaphore.acquire();
        System.out.println(res.q.take()+" "+res.atomicInteger.incrementAndGet());
        semaphore.release();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
};

作者:

Runnable writer = ()->{
    res.q.add("hi");
};

完整代码:

class Res{
    AtomicInteger atomicInteger = new AtomicInteger(0);
    BlockingQueue<String> q = new LinkedBlockingQueue<>();
}
public class Main {

    public static void main(String[] args) throws InterruptedException {

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
        Res res = new Res();
        Semaphore semaphore = new Semaphore(1); //this is for reader does not take two places in thread pool
        Runnable reader = ()->{
            try {
                semaphore.acquire();
                System.out.println(res.q.take()+" "+res.atomicInteger.incrementAndGet());
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        Runnable writer = ()->{
            res.q.add("hi");
        };
        Random rnd = new Random();

        for (int i = 0; i < 20; i++) {
            int time = rnd.nextInt(5)+ 2;
            executorService.schedule(writer,time, TimeUnit.SECONDS);
        }
        for (int i = 0; i < 20; i++) {
            int time = rnd.nextInt(5)+ 2;
            executorService.schedule(reader,time, TimeUnit.SECONDS);
        }

        executorService.shutdown();
}

它应该打印二十行“hi [number]”，但在某些行上卡住。 例如，我当前的打印:

hi 1
hi 2
hi 3
hi 4
hi 5

我发现如果我增加线程数newScheduledThreadPool(20)它就会开始工作，但是我怎样才能用两个线程来完成它呢？谢谢!

Answer 1

尽管很明显同时发生了什么，但遵循您的代码有点困难。由于 Executors.newScheduledThreadPool(2);，您最多可以一次运行两个线程。这两个线程都是读取器线程。

因此Thread-1进入了try block 并通过semaphore.acquire();获取了信号量许可，但是队列为空- 因此它会阻塞 res.q.take()。下一个线程 - Thread-2 也是一个读取线程，但它无法获取 permit，因为它已被 Thread-1 占用> 并在 semaphore.acquire(); 上被阻止。由于您没有空间容纳其他线程(您的池在使用这两个线程时被阻止)，因此没有编写器会将某些内容放入您的队列中，从而解锁 Thread-1 (因此 res.q.take() 可以工作)。

添加更多工作线程只会延迟问题的发生 - 您最终可能会处于与之前相同的位置。