有界优先阻塞队列

实际上我不会把它分类。虽然我现在不能把示例代码放在一起,但我建议使用decorator模式的一个版本。

PriorityBlockingQueue . 我发现这个类使用了以下接口:

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

在类的构造函数中,接受 PriorityBlockingQueue 作为构造函数参数。

然后通过 PriorityblockingQueue . 添加使其有界所需的任何代码。这是装饰器模式的一个相当标准的实现。

有一个在 Google Collections/Guava 图书馆: MinMaxPriorityQueue .

最小-最大优先级队列可以配置为最大大小。如果是的话, 每次队列大小超过该值时,队列 根据比较器自动删除其最大元素 来自传统的有界队列,这些队列阻塞或拒绝新的 充满时的元素。

在我的头顶上,我会将其子类化并覆盖put方法来强制执行此操作。如果发生异常,则抛出异常或执行任何适当的操作。

比如:

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:add和put invoke提供,所以覆盖它就足够了

根据Frank V的建议实现了BoundedPriorityBlockingQueue之后,我意识到它并没有达到我想要的效果。 主要的问题是,我必须插入到队列中的项可能比队列中已经存在的所有项具有更高的优先级。

为了充实弗兰克五世的建议,我使用了以下片段。。。

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

... 私有最终可重入锁定;//=新建ReentrantLock(); 私人最终条件未满;

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();
    }
}
...

我仍然在“PivotPriorityBlockingQueue”上工作,如果有人感兴趣,我可以发布它。

the ConcurrencyUtils repo

如果要执行的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);
    }
}

如果顺序需要严格,这显然会失败,因为耗尽将导致一个时刻,使用并发访问从队列中检索低优先级任务。

优点是,这是线程安全的,并且可能会以优先级队列设计所能达到的最快速度实现。

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

不幸的是,没有 阻塞队列 在标准Java库中实现。您要么需要找到一个实现,要么自己实现。实施 阻塞队列 需要一些正确锁定的知识。

https://gist.github.com/JensRantil/30f812dd237039257a3d 并使用它作为模板来实现您自己的包装器 SortedSet . 基本上,所有的锁都在那里,并且有多个单元测试(这需要一些调整)。

看一看这个 ForwardingQueue Semaphore .

还有另一个实现 here

它似乎在满足你的要求:

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