java - 带有 isEmpty() 更改通知的队列

Question

我有一个BlockingQueue<Runnable> (取自 ScheduledThreadPoolExecutor )在生产者-消费者环境中。有一个线程将任务添加到队列中，还有一个线程池执行这些任务。

我需要有关两个事件的通知:

1. 第一个项目已添加到空队列
2. 最后一项从队列中移除

通知 = 将消息写入数据库。

有什么明智的方法来实现这一点吗？

Answer 1

一个简单而幼稚的方法是使用一个实现来装饰您的 BlockingQueue，该实现只需检查底层队列，然后发布一个任务来执行通知。

NotifyingQueue<T> extends ForwardingBlockingQueue<T> implements BlockingQueue<T> {
  private final Notifier notifier; // injected not null

  …

  @Override public void put(T element) {
    if (getDelegate().isEmpty()) {
      notifier.notEmptyAnymore();
    }
    super.put(element);
  }

  @Override public T poll() {
    final T result = super.poll();
    if ((result != null) && getDelegate().isEmpty())
      notifier.nowEmpty();
  }
  … etc
}

这种方法有一些问题。虽然empty -> notEmpty 非常简单 - 特别是对于单个生产者的情况，但两个消费者很容易同时运行并且都看到队列从非空 -> 空。

如果您想要的只是通知队列在某个时间变空，那么只要您的通知程序是您的状态机，跟踪空性和非空性并在其从一种变为另一种时发出通知，就足够了:

AtomicStateNotifier implements Notifier {
  private final AtomicBoolean empty = new AtomicBoolean(true); // assume it starts empty
  private final Notifier delegate; // injected not null

  public void notEmptyAnymore() {
    if (empty.get() && empty.compareAndSet(true, false))
      delegate.notEmptyAnymore();
  }

  public void nowEmpty() {
    if (!empty.get() && empty.compareAndSet(false, true))
      delegate.nowEmpty();
  }
}

现在，这是围绕实际通知程序实现的线程安全防护，可能会将任务发布到执行程序以将事件异步写入数据库。