消费者线程等待多个队列

Question

我有多个任务队列和一个 consumer 线程。

Consumer 线程应该在任何队列有任务时唤醒。所以我想知道进行这种交流的最佳方式是什么。

这些是可以解决这个问题的一些实现以及为什么我想找到不同的东西的解释：

一种实现方法是使用一些监视器对象并调用 .wait(timeout) 从消费者线程对其调用 .notify() 来自生产者线程。然而，这种方法使用 wait/notify 哪个有点低级api，所以我尽可能避免使用它。而且它并不总是正确的，在某些情况下我们可能会等待整个超时，而我们有任务要做（睡觉的理发师问题）。
像 CountDownLatch 和 reset 方法会很好，但我在 java.util.concurrent 中没有找到类似的东西。实施会相当简单，但是实施新自行车是我比 wait/notify 更想避免的事情。我也相信它在等待整个超时时与 wait/notify 方法有同样的问题。
让生产者将创建的实体包装到一些 Task 中，并让所有生产者写入同一个队列，这样消费者就会监听一个队列。我相信这种方法在大多数情况下实际上非常好，但在我的情况下，这部分应用程序具有低延迟要求，因此我必须避免创建新对象（例如这些包装器），而且它会增加队列尾部的争用（而不是一个消费者，他们都会在那里写）这对延迟也不是很好。

那么还有其他实现方法吗？（可能使用其他一些并发原语）

Answer 1

如何使用任务通知队列，在这种情况下，如果任何任务队列添加项目，它也会将项目添加到通知队列。

以下片段说明了这种方法：

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

public class Main<T1, T2>  {


  Queue<T1> taskType1Queue = new ArrayBlockingQueue<T1>(10);
  Queue<T2> taskType2Queue= new ArrayBlockingQueue<T2>(10);
  ArrayBlockingQueue<Boolean> notificationQueue= new ArrayBlockingQueue<Boolean>(2);


  public void produceType1(T1 task) {
    new Thread(new Runnable() {
      @Override
      public void run() {
        taskType1Queue.add(task);
        notificationQueue.offer(true);; //does not block if full
      }
    });
  }

  public void produceType2(T2 task) {
    new Thread(new Runnable() {
      @Override
      public void run() {
        taskType2Queue.add(task);
        notificationQueue.offer(true); //does not block if full
      }
    });
  }


  public void consume() {

    try {
      notificationQueue.take();//wait till task1 o task2 has been published

      for(;!Thread.currentThread().isInterrupted();){
        T1 task1 = taskType1Queue.poll();//does not block if queue is empty
        if (task1 != null) {
          //do something
        }
        T2 task2 = taskType2Queue.poll();//does not block if queue is empty
        if (task2 != null) {
          //do something
        }
        if(task1 == null && task2 == null) {
          break;
        }
      }

    } catch (InterruptedException e) {
      System.out.println("Consumer thread done");
      return;
    }

  }

}

消费者线程等待多个队列

Consumer thread waiting on multiple qeueues

java

multithreading

producer-consumer