使用什么阻塞对象？

Question

我需要一个阻塞对象来触发一些事件。 （单个）消费者应该等待触发发生。然后它会做一些事情。然后它再次等待触发。 触发器由多个不同的线程（生产者）激活。但是生产者不产生任何数据。这种触发器的语义是：“消费者必须做某事”（例如重新计算一些值，因为底层数据发生了变化）。 这意味着即使触发器被多次激活，它也应该作为单个触发器出现在消费者面前。

我考虑过使用CountDownLatch 或ArrayBlockingQueue，但它们似乎不合适。

这是我想使用的触发器构造：

public class Trigger{
  private final MagicBlockingObject blockingLatch;

  public void trigger(){
     //activate the blockingLatch, so that a call to waitForTrigger() returns
  } 

  public void waitForTrigger(){
    //read from the blockingLatch. This should block until trigger() is called.
  }
}

对MagicBlockingObject 使用什么有任何想法？

BlockingQueue 似乎合适，但如果队列已满，我没有找到将其限制为单个内容对象而不阻塞生产者的方法。

Answer 1

您可以使用容量为 1 的 ArrayBlockingQueue 来解决此问题：

public class Trigger{
  private final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

  public void trigger(){
     queue.offer("foo");
  } 

  public void waitForTrigger(){
    queue.take();
  }
}

Answer 2

这样的简单解决方案有什么问题：

public class Trigger {
    private final Object blockingLatch = new Object();

    public void trigger() {
        //activate the blockingLatch, so that a call to waitForTrigger() returns
        synchronized(blockingLatch){
            blockingLatch.notify();
        }
    }

    public void waitForTrigger() throws InterruptedException {
        //read from the blockingLatch. This should block until trigger() is called.
        synchronized(blockingLatch){
            blockingLatch.wait();
        }
    }
}

消费者将调用 waitForTrigger() 并阻塞直到生产者没有调用 trigger()。如果消费者没有被阻塞，那么生产者调用 trigger() 不会有任何影响。

Answer 3

java.util.concurrent 有很多不错的实用程序。 wait 和 notify 应该被认为是过时的。

如果我了解您的问题，您可以尝试使用Semaphore

public class Blocking {
    private final Semaphore openTasks = new Semaphore(0);
    public void addTask() {
        // add 1
        openTasks.release();
    }
    public void takeAllTasks() throws InterruptedException {
        int immediately = openTasks.drainPermits();
        if (immediately > 0) {
            // there was a task, no need to wait
            return;
        }
        // wait for task
        openTasks.acquire();
        // ensure we leave this method without leaving permits
        openTasks.drainPermits()
    }
}

无限数量的生产者可以向信号量添加“许可”，而您的消费者只需将它们全部拿走或等待至少一个出现。