java 有界优先阻塞队列

Bounded PriorityBlockingQueue

提问by nanda

PriorityBlockingQueueis unbounded, but I need to bound it somehow. What is the best way to achieve that?

PriorityBlockingQueue是无限的，但我需要以某种方式绑定它。实现这一目标的最佳方法是什么？

For information, the bounded PriorityBlockingQueuewill be used in a ThreadPoolExecutor.

有关信息，有界PriorityBlockingQueue将用于ThreadPoolExecutor.

NB: By bounded I don't want to throw Exception if that happens, I want to put the object in the queue and then cut it based on its priority value. Is there any good way to do this cut thingie?

注意：如果发生这种情况，我不想抛出异常，我想将对象放入队列，然后根据其优先级值剪切它。有没有什么好方法可以做到这一点？

采纳答案by Frank V

I actually wouldn't subclass it. While I can't put together example code right now, I'd suggest a version of the decorator pattern.

我实际上不会对它进行子类化。虽然我现在无法将示例代码放在一起，但我建议使用装饰器模式的一个版本。

Create a new class and implement the interfaces implemented by your class of interest: PriorityBlockingQueue. I've found the following interfaces used by this class:

创建一个新类并实现您感兴趣的类实现的接口：PriorityBlockingQueue。我发现这个类使用了以下接口：

Serializable, Iterable<E>, Collection<E>, BlockingQueue<E>, Queue<E>

In the constructor for a class, accept a PriorityBlockingQueueas a constructor parameter.

在类的构造函数中，接受 aPriorityBlockingQueue作为构造函数参数。

Then implement all the methods required by the interfaces via the instances of the PriorityblockingQueue. Add any code required to make it Bounded.

然后通过PriorityblockingQueue. 添加使其有界所需的任何代码。

Here's a quick diagram I put together in Violet UML:

这是我在Violet UML 中整理的快速图表：

BoundedPriorityblockingQueue class diagram http://theopensourceu.com/wp-content/uploads/2010/03/BoundedPriorityBlockingQueue.png

BoundedPriorityblockingQueue 类图 http://theopensourceu.com/wp-content/uploads/2010/03/BoundedPriorityBlockingQueue.png

回答by rescdsk

There's an implementation of this in the Google Collections/Guavalibrary: MinMaxPriorityQueue.

在Google Collections/Guava库中有一个实现：MinMaxPriorityQueue。

A min-max priority queue can be configured with a maximum size. If so, each time the size of the queue exceeds that value, the queue automatically removes its greatest element according to its comparator (which might be the element that was just added). This is different from conventional bounded queues, which either block or reject new elements when full.

min-max 优先级队列可以配置为最大大小。如果是这样，每次队列的大小超过该值时，队列会根据其比较器（可能是刚刚添加的元素）自动删除其最大元素。这与传统的有界队列不同，后者在满时阻塞或拒绝新元素。

回答by Kris

Of the top of my head, I'd subclass it and overwrite the put method to enforce this. If it goes over throw an exception or do whatever seems appropriate.

在我的头顶上，我会将它子类化并覆盖 put 方法以强制执行此操作。如果它超过抛出异常或做任何看起来合适的事情。

Something like:

就像是：

public class LimitedPBQ extends PriorityBlockingQueue {

    private int maxItems;
    public LimitedPBQ(int maxItems){
        this.maxItems = maxItems;
    }

    @Override
    public boolean offer(Object e) {
        boolean success = super.offer(e);
        if(!success){
            return false;
        } else if (this.size()>maxItems){
            // Need to drop last item in queue
            // The array is not guaranteed to be in order, 
            // so you should sort it to be sure, even though Sun's Java 6 
            // version will return it in order
            this.remove(this.toArray()[this.size()-1]);
        }
        return true;
    }
}

Edit: Both add and put invoke offer, so overriding it should be enough

编辑：添加和放置调用报价，所以覆盖它应该就足够了

Edit 2: Should now remove the last element if over maxItems. There may be a more elegant way of doing it though.

编辑 2：如果超过 maxItems，现在应该删除最后一个元素。不过，可能有一种更优雅的方式来做到这一点。

回答by phreed

After implementing a BoundedPriorityBlockingQueue according to what Frank V suggested I realized it didn't do quite what I wanted. The main problem is that the item which I have to insert into the queue may be a higher priority than everything already in the queue. Thus what I really want is a 'pivot' method, if I put an object into the queue, when the queue is full, I want to get back the lowest priority object, rather than blocking.

根据 Frank V 的建议实现了 BoundedPriorityBlockingQueue 后，我意识到它并没有达到我想要的效果。主要问题是我必须插入队列的项目可能比队列中已有的所有项目具有更高的优先级。因此，我真正想要的是一个“枢轴”方法，如果我将一个对象放入队列中，当队列已满时，我想取回优先级最低的对象，而不是阻塞。

To flesh out Frank V's suggestions I used the following fragments...

为了充实 Frank V 的建议，我使用了以下片段......

public class BoundedPriorityBlockingQueue<E> 
   implements 
     Serializable, 
     Iterable<E>, 
     Collection<E>, 
     BlockingQueue<E>, 
     Queue<E>, 
     InstrumentedQueue 
{

... private final ReentrantLock lock; // = new ReentrantLock(); private final Condition notFull;

... 私有最终 ReentrantLock 锁；// = 新的 ReentrantLock(); 私人最终条件 notFull;

final private int capacity;
final private PriorityBlockingQueue<E> queue;

public BoundedPriorityBlockingQueue(int capacity) 
  throws IllegalArgumentException, 
         NoSuchFieldException, 
         IllegalAccessException 
{
   if (capacity < 1) throw 
       new IllegalArgumentException("capacity must be greater than zero");      
   this.capacity = capacity;
   this.queue = new PriorityBlockingQueue<E>();

   // gaining access to private field
   Field reqField;
   try {
    reqField = PriorityBlockingQueue.class.getDeclaredField("lock");
    reqField.setAccessible(true);
    this.lock = (ReentrantLock)reqField.get(ReentrantLock.class);
    this.notFull = this.lock.newCondition();

   } catch (SecurityException ex) {
    ex.printStackTrace();
    throw ex;
   } catch (NoSuchFieldException ex) {
    ex.printStackTrace();
    throw ex;
   } catch (IllegalAccessException ex) {
    ex.printStackTrace();
    throw ex;
   }

...

@Override
public boolean offer(E e) {
    this.lock.lock();
    try {
        while (this.size() == this.capacity)
            notFull.await();
        boolean success = this.queue.offer(e);
        return success;
    } catch (InterruptedException ie) {
        notFull.signal(); // propagate to a non-interrupted thread
        return false;

    } finally {
        this.lock.unlock();
    }
}
...

This also has some instrumentation so I can check the effectiveness of the queue. I am still working on 'PivotPriorityBlockingQueue', if anyone is interested I can post it.

这也有一些工具，所以我可以检查队列的有效性。我仍在研究“PivotPriorityBlockingQueue”，如果有人感兴趣，我可以发布它。

回答by Christopher Oezbek

If the order of the Runnables you want to execute is not strict (as is: it may occur that some lower priority tasks are executed even though higher priority tasks exist), then I would suggest the following, which boils down to periodically cutting the PriorityQueue down in size:

如果您要执行的 Runnables 的顺序不严格（原样：即使存在较高优先级的任务，也可能会执行一些较低优先级的任务），那么我建议以下，归结为定期切割 PriorityQueue缩小尺寸：

if (queue.size() > MIN_RETAIN * 2){
    ArrayList<T> toRetain = new ArrayList<T>(MIN_RETAIN);
    queue.drainTo(toRetain, MIN_RETAIN);
    queue.clear();
    for (T t : toRetain){
      queue.offer(t);
    }
}

This will obviously fail if the order needs to be strict, as draining will lead to a moment, wenn low priority task will retrieved from the queue using concurrent access.

如果顺序需要严格，这显然会失败，因为排空将导致片刻，wenn 低优先级任务将使用并发访问从队列中检索。

The advantages are, that this is thread-safe and likely to get as fast as you can do with the priority queue design.

优点是，这是线程安全的，并且可能与优先级队列设计一样快。

回答by Ztyx

Not a single answer so far has all of the following properties:

到目前为止，没有一个答案具有以下所有属性：

• Implements the BlockingQueueinterface.
• Supports removal of the absolute largest value.
• No race conditions.

• 实现BlockingQueue接口。
• 支持去除绝对最大值。
• 没有比赛条件。

Unfortunately, there is no BlockingQueueimplementation in the standard Java library. You will either need to find an implementation or implement something yourself. Implementing a BlockingQueuewill require some knowledge on proper locking.

不幸的是，BlockingQueue标准 Java 库中没有实现。你要么需要找到一个实现，要么自己实现一些东西。实现 a BlockingQueuewill 需要一些有关正确锁定的知识。

Here's what I suggest: Have a look at https://gist.github.com/JensRantil/30f812dd237039257a3dand use it as a template to implement your own wrapper around a SortedSet. Basically, all the locking is there and there are multiple unit tests (that will need some tweaking).

这是我的建议：查看https://gist.github.com/JensRantil/30f812dd237039257a3d并将其用作模板来实现您自己的SortedSet. 基本上，所有锁定都在那里，并且有多个单元测试（需要一些调整）。

回答by bryant1410

There is an implementation in the ConcurrencyUtils repo.

ConcurrencyUtils repo 中有一个实现。

回答by AbuZubair

There is another implementation here

还有另一种实现在这里

It seems to do what you are asking for:

它似乎可以满足您的要求：

A BoundedPriorityQueue implements a priority queue with an upper bound on the number of elements. If the queue is not full, added elements are always added. If the queue is full and the added element is greater than the smallest element in the queue, the smallest element is removed and the new element is added. If the queue is full and the added element is not greater than the smallest element in the queue, the new element is not added.

BoundedPriorityQueue 实现了一个具有元素数量上限的优先级队列。如果队列未满，则始终添加添加的元素。如果队列已满并且添加的元素大于队列中的最小元素，则移除最小元素并添加新元素。如果队列已满且添加的元素不大于队列中的最小元素，则不添加新元素。

回答by yawn

Have a look at the ForwardingQueuefrom the Google Collections API. For blocking semantics you could use a Semaphore.

查看来自 Google Collections API的ForwardingQueue。对于阻塞语义，您可以使用Semaphore。