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

Question

我在生产者-消费者环境中有一个 BlockingQueue（取自 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
}

但这种方法有几个问题。而空-> 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();
  }
}

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

Answer 2

该设计很可能有缺陷，但您可以相对简单地做到这一点：
您有一个单线程添加，因此您可以在添加之前进行检查。即 pool.getQueue().isEmpty() - 只有一个生产者，这是安全的。

无法保证最后删除的项目，但您可以覆盖 beforeExecute 并再次检查队列。在 isEmpty() 返回 true 后可能会有一个小超时。也许下面的代码在 afterExecute 中执行会更好。

protected void beforeExecute(Thread t, Runnable r) { 
  if (getQueue().isEmpty()){
    try{
      Runnable r = getQueue().poll(200, TimeUnit.MILLISECONDS);
      if (r!=null){ 
        execute(r);
      }  else{
        //last message - or on after execute by Setting a threadLocal and check it there
        //alternatively you may need to do so ONLY in after execute, depending on your needs
     }
    }catch(InterruptedException _ie){
      Thread.currentThread().interrupt();
    }
  }
}

有时

我可以解释为什么使用队列本身进行通知不能很好地工作：假设您添加一个要由池执行的任务，该任务立即计划，队列再次为空，您将需要通知。