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

它应该打印二十行“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() 就可以工作了)。

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