Java - BlockingQueue冻结多线程应用

Question

我正在制作一个包含两个线程的应用程序:一个线程向LinkedBlockingQueue写入值，另一个线程读取数据。我使用ScheduledExecutorService在一段时间内以秒为单位运行此操作。问题是我的应用程序在BlockingQueue的方法上冻结了，我不明白为什么。

这是一个常见的资源：

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();
}

它应该打印20行"hi number"，但在某些行处冻结。例如，我当前的打印：

hi 1
hi 2
hi 3
hi 4
hi 5

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

Answer 1

要理解你的代码有点困难，尽管很明显正在发生的事情。由于使用了Executors.newScheduledThreadPool(2);，您最多只能同时运行两个线程。这两个线程都是reader线程。

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

添加更多的工作线程只会延迟问题--你可能会陷入和以前一样的境地。