我有一个PriorityBlockingQueue。单个线程一次使用此队列中的一条消息并对其进行处理。其他几个线程正在将消息插入队列。生产者线程为他们提交的每条消息分配一个完整的优先级。静态AtomicLong 用于为每条消息分配一个唯一的、单调递增的ID。队列的Comparator先按此优先级对消息进行排序，然后同等优先级的消息按ID排序（ID最低的在前）。
问题：有时一个生产者会提交大量消息。然后，这会使其他生产者无法处理他们的消息。我想做的是在生产者之间进行消费者循环以获取相同优先级的消息（同时仍按提交顺序处理单个生产者的同等优先级消息）。但我不知道如何编写 Comparator 来做到这一点。
我考虑的另一种选择是为每个生产者设置一个单独的队列。但是，我认为这行不通，因为我不知道单线程在多个队列上等待的任何方式。
【问题讨论】：
标签： java multithreading priority-queue java.util.concurrent round-robin
我觉得为每个生产者使用一个Queue 来实现这一点更简单。一个线程不能等待多个Queues，但您可以将所有Queues 合并到一个辅助类中，这样就不需要了。
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.annotation.concurrent.GuardedBy;
public class RoundRobin<P, E> {
private final Lock lock = new ReentrantLock();
private final Condition added = lock.newCondition();
@GuardedBy("lock") private final Map<P, Queue<E>> queues = new LinkedHashMap<>();
public boolean add(P producer, E item) {
lock.lock();
try {
if (!queues.containsKey(producer)) {
queues.put(producer, new PriorityBlockingQueue<>());
}
added.signalAll();
return queues.get(producer).add(item);
} finally {
lock.unlock();
}
}
public Iterator<E> roundRobinIterator() {
return new Iterator<E>() {
private Iterator<? extends Queue<E>> i = null;
private boolean singlePass = true;
@Override
public boolean hasNext() {
return true;
}
@Override
public E next() {
lock.lock();
try {
while (true) {
if (i == null || !i.hasNext()) {
i = queues.values().iterator();
singlePass = true;
}
while (i.hasNext()) {
Queue<E> q = i.next();
if (!q.isEmpty()) {
if (singlePass) {
// copy the iterator to prevent
// ConcurrentModificationExceptions
singlePass = false;
i = copy(i);
}
return q.poll();
}
}
if (singlePass) {
// If singlePass is true then we just checked every
// queue and they were all empty.
// Wait for another element to be added.
added.await();
}
}
} catch (InterruptedException e) {
throw new NoSuchElementException(e.getMessage());
} finally {
lock.unlock();
}
}
private <T> Iterator<? extends T> copy(Iterator<? extends T> i) {
List<T> copy = new ArrayList<>();
while (i.hasNext()) {
copy.add(i.next());
}
return copy.iterator();
}
};
}
}
【讨论】：
我想我会这样做：
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
public class RRQueue<M> {
private final ThreadLocal<Queue<M>> threadQueue = new ThreadLocal<>();
private final List<Queue<M>> queues;
private int current = 0;
public RRQueue() {
this.queues = new ArrayList<>();
}
public synchronized void add(M msg) {
Queue<M> queue = threadQueue.get();
if (queue == null) {
queue = new LinkedList<>(); // or whatever
queues.add(queue);
threadQueue.set(queue);
}
queue.add(msg);
notify();
}
public synchronized M get() throws InterruptedException {
while (true) {
for (int i = 0; i < queues.size(); ++i) {
Queue<M> queue = queues.get(current);
current = (current+1)%queues.size();
if (!queue.isEmpty()) {
return queue.remove();
}
}
wait();
}
}
}
【讨论】：
这完全取决于您如何分配 ID。将它们分开分配 N，其中 N 是生产者的数量，并将每个生产者的索引添加到其中。然后按顺序读取它们将产生循环顺序。您需要做一点记账才能知道何时增加底层 ID，这将在您达到任何 x 的 Nx-1 时发生。
【讨论】：
可以使用 PriorityBlockingQueue 实现 N 个生产者之间的循环，如下所示。
为每个生产者维护 N 个 AtomicInteger 计数器和一个全局计数器，以防生产者计数器相等。
当添加到生产者的 Q incement 计数器以及全局计数器并存储到 Q 对象中时。 Q 的比较器将根据生产者计数器 1st 排序，然后是存储在 Q 对象中的全局计数器值。
但是，当一个生产者的对象在 Q 中的某个时间为空时，相应的计数器就会落后，并且当对象开始进入时它将开始占用 Q。
为避免这种情况，请维护一个 volatile 变量，该变量在出队时使用对象的生产者计数器进行更新。如果值小于 volatile 变量中最后一个出队的计数器，则在入队期间增加生产者计数器后，将计数器重置为该值 + 1。
【讨论】：