阻塞和唤醒消费者线程/

Question

我有 2 个线程。

他们每个人都从共享缓冲区中读取一些数据。

currentDataBuffer.get(thisID); //currentDataBuffer is my shared buffer object

我想在每次调用 get 后阻塞每个线程，并在所有线程读取缓冲区时释放它（一次） 所以我用这个currentDataBuffer 对象作为锁：

currentDataBuffer.get(thisID);
synchronized (currentDataBuffer ) {
   currentDataBuffer.wait();
}

问题是当所有线程完成从缓冲区读取（每个线程）时，我如何释放这些线程？

在currentDataBuffer 内部，我有一个映射，其中存储了从缓冲区读取数据的线程的 ID。

如何使用this.notifyAll();（来自currentDataBuffer）唤醒所有锁定的线程？

Answer 1

我建议你使用 Java BlockingQueue 数据结构。调用BlockingQueue.take() 会阻塞直到元素可用。 所以而不是：

currentDataBuffer.get(thisID);
synchronized (currentDataBuffer ) {
   currentDataBuffer.wait();
}

你将拥有：

currentDataBuffer.take();

在take()调用中阻塞的线程可以通过调用BlockingQueue.offer(object)方法向队列中添加一个元素来释放

Answer 2

我认为 CountdownLatch (http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/CountDownLatch.html) 会有所帮助。

即使用初始值为 2 的 CountDownLatch （或您要处理的任何线程数）增加您的 currentDataBuffer。处理完线程调用latch.countDown()然后latch.await()。

为了安全起见，您可能需要小心 countDowns 不会丢失（例如，如果抛出异常）。

Answer 3

继续您的代码的答案可能是使用currentDataBuffer.notifyAll() 而不是this.notifyAll()（不清楚this 从您的问题中指的是什么）。但是，您如何确保在所有线程都读取缓冲区并进入等待状态之后调用notifyAll？

如果您知道应该读取缓冲区的线程数，更好的解决方案是使用两个CountDownLatch。该类的 javadoc 中有一个很好的示例。

------------ 更新：

您无需更改阅读器线程的代码。您应该在 currentDataBuffer 对象中握住闩锁。在currentDataBuffer 中创建一个等待n 项目锁存器的线程。在get 方法中，在读取完成后对该锁存器调用CountDownLatch.countDown()。哦，我会写的：

class CurrentDataBuffer implements Runnable {

    private CountDownLatch singleThreadFinishedSignal;
    private CountDownLatch allThreadsFinishedSignal;

    public CurrentDataBuffer(int N) {
        singleThreadFinishedSignal = new CountDownLatch(N); // waiter thread waits on this
        allThreadsFinishedSignal = new CountDownLatch(1); // reader threads wait on this after finished reading
    }

    public void run() {
        try {
            singleThreadFinishedSignal.await(); // wait for all reader threads to finish
            allThreadsFinishedSignal.countDown(); // let all reader threads proceed
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void get() {
        try {
            // read buffer item here ///////////////////
            singleThreadFinishedSignal.countDown(); // mark that a thread has read the item
            allThreadsFinishedSignal.await(); // wait for other reader threads to finish
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}