java - 如何在多个阻塞队列上等待？

Question

摘要

我有一个消费者，多个生产者问题，每个生产者将其数据放在单独的队列上(因为它们的数据类型不同)，并且生产者都在自己的线程上。

消费者将在循环中根据优先级从生产者队列中提取项目，然后当所有队列为空时，它应该进入休眠状态，直到产生更多数据。

我不知道让消费者进入休眠状态然后再唤醒它的最佳(无错误和最快)方法。

当前代码

目前我正在使用一个简单的信号量。每次产生新数据时，生产者都会调用:

void produce(Data data) {
    queue.add(data);
    semaphore.release();
}

消费者:

while (!Thread.currentThread().isInterrupted()) {
    consumeQueues();
    semaphore.tryAcquire(time, TimeUnit.SECONDS);
    semaphore.drainPermits();
}

当前想法

使用空的选择器，让消费者调用selector.select()，生产者调用selector.wakeup()。

重写所有生成的值并切换到优先级队列。这将需要大量工作，而且我什至不确定我是否可以在不损失速度的情况下获得与现在完全相同的行为。

使用lock/conditional方法并调用conditional.await()和conditional.signal() 。我担心添加大量代码并因添加锁获取(性能敏感代码)而减慢速度。

最好我只使用一个静态 AtomicBoolean，当数据可用时，生产者可以将其延迟设置为 true，但是 AtomicBoolean 没有阻塞，直到-真正的方法。

Answer 1

您可以使用以下结构:阻塞队列的优先级队列(如果同一类型内有优先级，则为优先级队列)

你应该做一个像ProducedItem这样的空接口(interface)，所有创建的项目都应该实现它，这样它就可以编译。

PriorityBlockingQueue<PriorityBlockingQueue<ProducedItem>> queue = new PriorityBlockingQueue<PriorityBlockingQueue<ProducedItem>>;

然后，当您添加新生产的项目时:

void addWorkItem(int queueIndex, ProducedItem producedItem) {
    //Note: You may want to make the two operations an atomic operation
    priorityQueues[queueIndex].add(workItem);
    prioritypriorityQueue.add(producers[queueIndex]);
}