是否有具有固定容量和自定义比较器的 PriorityQueue 实现？答案

【问题标题】：Is there a PriorityQueue implementation with fixed capacity and custom comparator?是否有具有固定容量和自定义比较器的 PriorityQueue 实现？
【发布时间】：2011-12-14 05:59:00
【问题描述】：

相关问题：

我有一个非常大的数据集（超过 500 万个项目），我需要从中获取 N 个最大的个项目。最自然的方法是使用堆/优先队列只存储前 N 个项目。 JVM（Scala/Java）的优先级队列有几个很好的实现，分别是：

前 2 个很好，但它们存储了所有项目，在我的情况下，这会产生严重的内存开销。第三个（Lucene 实现）没有这样的缺点，但正如我从文档中看到的那样，它也不支持自定义比较器，这对我来说毫无用处。

所以，我的问题是：是否有 PriorityQueue 实现具有固定容量和自定义比较器？

UPD。最后我根据彼得的回答创建了自己的实现：

public class FixedSizePriorityQueue<E> extends TreeSet<E> {

    private int elementsLeft;

    public FixedSizePriorityQueue(int maxSize) {
        super(new NaturalComparator());
        this.elementsLeft = maxSize;
    }

    public FixedSizePriorityQueue(int maxSize, Comparator<E> comparator) {
        super(comparator);
        this.elementsLeft = maxSize;
    }


    /**
     * @return true if element was added, false otherwise
     * */
    @Override
    public boolean add(E e) {
        if (elementsLeft == 0 && size() == 0) {
            // max size was initiated to zero => just return false
            return false;
        } else if (elementsLeft > 0) {
            // queue isn't full => add element and decrement elementsLeft
            boolean added = super.add(e);
            if (added) {
                elementsLeft--;
            }
            return added;
        } else {
            // there is already 1 or more elements => compare to the least
            int compared = super.comparator().compare(e, this.first());
            if (compared == 1) {
                // new element is larger than the least in queue => pull the least and add new one to queue
                pollFirst();
                super.add(e);
                return true;
            } else {
                // new element is less than the least in queue => return false
                return false;
            }
        }
    }
}

（其中NaturalComparator 取自this 问题）

【问题讨论】：

我对你的实现的拙见：1）你真的需要扩展TreeSet吗？ “FixedSizePriorityQueue 是一个 TreeSet”听起来不太好，我会让该集合成为一个成员。 2）您实际上并不需要通过使 elementsLeft 成为非最终变量来为您的类添加状态。 3) 你确定 add 方法总是返回正确的值吗？ 4) 防止无效或非法参数是一种好习惯。
@Murat：感谢您的建议。随时在此处发布您改进的实现作为答案。
不客气，谢谢:) 将其发布为答案。
如果队列已满并且您添加了一个已经存在于其中的元素（并且不是将被轮询的最后一个元素），您的解决方案将错误地离开 elementsLeft == 0，而在这种情况下它必须变成 1。
其实队列满的时候应该先添加元素并查看结果。如果为假，则不必投票。

标签： java scala heap priority-queue

【解决方案1】：

你怎么能说 Lucene 不支持自定义比较器？

它的抽象，你必须实现抽象方法lessThan(T a, T b)

【讨论】：

呵呵，没注意，谢谢！赞成，但由于它需要额外的库，我相信基于标准 API 的另一种实现更可取（请参阅我的更新）。

【解决方案2】：

您可以使用 SortedSet，例如带有自定义比较器的 TreeSet，并在大小达到 N 时删除最小的。

【讨论】：

TreeSet 在此用例中的性能将低于 PriorityQueue。 stackoverflow.com/questions/3524862/…
或者简单地说，对优先队列做同样的事情？编辑：我在下面添加了一个答案来说明我的意思。

【解决方案3】：

虽然是一个老问题，但它可能对其他人有帮助。您可以使用 Google 的 Java 库 guava 的minMaxPriorityQueue。

【讨论】：

仍标记为@Beta，但自 8.0 以来一直存在，所以我怀疑它相当可靠。
这实际上并不是 MinMaxPriorityQueue 的预期用途，它在该用例中表现不佳。
@LouisWasserman 为什么会这样？不就是O(n*log(k))（其中n是数据集大小，k是队列的最大大小”？
@Kranach 常数因子将明显比正常的PriorityQueue 差。使用普通的PriorityQueue 会做得更好，或者更好，Ordering.greatestOf 使用 O(n) 时间，O(k) 内存算法。（我们正在考虑弃用 MinMaxPriorityQueue，只是因为它往往会以这种方式被滥用。）

【解决方案4】：

我想不出一个现成的，但你可以检查my implementation这个集合有类似的要求。

区别在于比较器，但如果您从PriorityQueue 扩展，您将拥有它。并在每次添加时检查您是否未达到限制，如果已达到 - 删除最后一项。

【讨论】：

不幸的是，标准PriorityQueue 没有提供简单（快速）的方法来删除最少元素（这对于堆结构来说是可以理解的）。所以我决定在 TreeSet 之上实现固定大小的优先级队列。还是谢谢。
@Robert Muir: poll() 删除队列的头部，即最大元素，而不是最小元素。
@Robert Muir：呵呵，你又说对了！我想象了另一个堆实现（树状），甚至没有想过获得最少元素很容易，因此我确定head 是顶级元素并且在javadocs 中错过了这一点。现在我看到了。再次感谢！
topN 的习惯用法类似于 1. 如果 pq.size == N && item N), pq.poll()。 lucene 在这里提供了 2 个优点： 1. 如果 N 很小，则使用哨兵填充以避免大小检查。 2.如果item是可变的，而不是offer + poll，你只需改变head并调用updateTop()。

【解决方案5】：

以下是我之前使用的实现。遵从彼得的建议。

 public @interface NonThreadSafe {
 }

/**
 * A priority queue implementation with a fixed size based on a {@link TreeMap}.
 * The number of elements in the queue will be at most {@code maxSize}.
 * Once the number of elements in the queue reaches {@code maxSize}, trying to add a new element
 * will remove the greatest element in the queue if the new element is less than or equal to
 * the current greatest element. The queue will not be modified otherwise.
 */
@NonThreadSafe
public static class FixedSizePriorityQueue<E> {
    private final TreeSet<E> treeSet; /* backing data structure */
    private final Comparator<? super E> comparator;
    private final int maxSize;

    /**
     * Constructs a {@link FixedSizePriorityQueue} with the specified {@code maxSize}
     * and {@code comparator}.
     *
     * @param maxSize    - The maximum size the queue can reach, must be a positive integer.
     * @param comparator - The comparator to be used to compare the elements in the queue, must be non-null.
     */
    public FixedSizePriorityQueue(final int maxSize, final Comparator<? super E> comparator) {
        super();
        if (maxSize <= 0) {
            throw new IllegalArgumentException("maxSize = " + maxSize + "; expected a positive integer.");
        }
        if (comparator == null) {
            throw new NullPointerException("Comparator is null.");
        }
        this.treeSet = new TreeSet<E>(comparator);
        this.comparator = treeSet.comparator();
        this.maxSize = maxSize;
    }

    /**
     * Adds an element to the queue. If the queue contains {@code maxSize} elements, {@code e} will
     * be compared to the greatest element in the queue using {@code comparator}.
     * If {@code e} is less than or equal to the greatest element, that element will be removed and
     * {@code e} will be added instead. Otherwise, the queue will not be modified
     * and {@code e} will not be added.
     *
     * @param e - Element to be added, must be non-null.
     */
    public void add(final E e) {
        if (e == null) {
            throw new NullPointerException("e is null.");
        }
        if (maxSize <= treeSet.size()) {
            final E firstElm = treeSet.first();
            if (comparator.compare(e, firstElm) < 1) {
                return;
            } else {
                treeSet.pollFirst();
            }
        }
        treeSet.add(e);
    }

    /**
     * @return Returns a sorted view of the queue as a {@link Collections#unmodifiableList(java.util.List)}
     *         unmodifiableList.
     */
    public List<E> asList() {
        return Collections.unmodifiableList(new ArrayList<E>(treeSet));
    }
}

如果有任何反馈，我将不胜感激。

编辑： 使用TreeSet 似乎不是很有效，因为对first() 的调用似乎需要亚线性时间。我将TreeSet 更改为PriorityQueue。修改后的add() 方法如下所示：

   /**
     * Adds an element to the queue. If the queue contains {@code maxSize} elements, {@code e} will
     * be compared to the lowest element in the queue using {@code comparator}.
     * If {@code e} is greater than or equal to the lowest element, that element will be removed and
     * {@code e} will be added instead. Otherwise, the queue will not be modified
     * and {@code e} will not be added.
     *
     * @param e - Element to be added, must be non-null.
     */
    public void add(final E e) {
        if (e == null) {
            throw new NullPointerException("e is null.");
        }
        if (maxSize <= priorityQueue.size()) {
            final E firstElm = priorityQueue.peek();
            if (comparator.compare(e, firstElm) < 1) {
                return;
            } else {
                priorityQueue.poll();
            }
        }
        priorityQueue.add(e);
    }

【讨论】：

谢谢！ ...恕我直言，对于基于 PriorityQueue 的实现，asList() 方法应该类似于：List<E> mutableList = new ArrayList<E>(priorityQueue); Collections.sort(mutableList, comparator); return Collections.unmodifiableList( mutableList );
@Abdull 是对的。您的 javadoc 说它返回一个排序视图，但 PriorityQueue 的迭代器不保证元素按顺序返回。

【解决方案6】：

正是我想要的。但是，该实现包含一个错误：

即：如果elementsLeft > 0 并且e 已经包含在TreeSet 中。在这种情况下，elementsLeft 减少了，但 TreeSet 中的元素数量保持不变。

我建议将 add() 方法中的相应行替换为

        } else if (elementsLeft > 0) {
        // queue isn't full => add element and decrement elementsLeft
        boolean added = super.add(e);
        if (added) {
            elementsLeft--;
        }
        return added;

【讨论】：

【解决方案7】：

试试这个代码：

public class BoundedPQueue<E extends Comparable<E>> {
/**
 * Lock used for all public operations
 */
private final ReentrantLock lock;

PriorityBlockingQueue<E> queue ;
int size = 0;

public BoundedPQueue(int capacity){
    queue = new PriorityBlockingQueue<E>(capacity, new CustomComparator<E>());
    size = capacity;
    this.lock = new ReentrantLock();

}

public boolean offer(E e) {


    final ReentrantLock lock = this.lock;
    lock.lock();
    E vl = null;
    if(queue.size()>= size)  {
        vl= queue.poll();
        if(vl.compareTo(e)<0)
            e=vl;
    }

    try {
        return queue.offer(e);
    } finally {
        lock.unlock();
    }


}

public E poll()  {

    return queue.poll();
}

public static class CustomComparator<E extends Comparable<E>> implements Comparator<E> {


    @Override
    public int compare(E o1, E o2) {
        //give me a max heap
         return o1.compareTo(o2) *-1;

    }
}

}

【讨论】：

【解决方案8】：

如果你有番石榴，这是我放在一起的一个。我认为它是相当完整的。如果我错过了什么，请告诉我。

您可以使用 gauva ForwardingBlockingQueue，这样您就不必映射所有其他方法。

import com.google.common.util.concurrent.ForwardingBlockingQueue;

public class PriorityBlockingQueueDecorator<E> extends
        ForwardingBlockingQueue<E> {

    public static final class QueueFullException extends IllegalStateException {

        private static final long serialVersionUID = -9218216017510478441L;

    }

    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

    private int maxSize;

    private PriorityBlockingQueue<E> delegate;

    public PriorityBlockingQueueDecorator(PriorityBlockingQueue<E> delegate) {
        this(MAX_ARRAY_SIZE, delegate);
    }

    public PriorityBlockingQueueDecorator(int maxSize,
            PriorityBlockingQueue<E> delegate) {
        this.maxSize = maxSize;
        this.delegate = delegate;
    }

    @Override
    protected BlockingQueue<E> delegate() {
        return delegate;
    }

    @Override
    public boolean add(E element) {
        return offer(element);
    }

    @Override
    public boolean addAll(Collection<? extends E> collection) {
        boolean modified = false;
        for (E e : collection)
            if (add(e))
                modified = true;
        return modified;
    }

    @Override
    public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
        return offer(e);
    }

    @Override
    public boolean offer(E o) {
        if (maxSize > size()) {
            throw new QueueFullException();
        }
        return super.offer(o);
    }
}

【讨论】：

【解决方案9】：

嗯，这是一个很老的问题，但我很困惑为什么还没有建议更简单的解决方案。

除非我遗漏了什么，否则这可以使用 min-heap（Java 的默认 PriorityQueue 实现） 轻松解决，但在 PriorityQueue 的大小变得大于 k 的那一刻稍有变化（即，如果我们尝试存储前 k 个元素），则轮询头部。

这是我的意思的一个例子

public void storeKLargest(int[] nums, int k) {
    PriorityQueue<Integer> pq = new PriorityQueue<>(k+1);
    for(int num: nums){
        if(pq.size() < k || pq.peek() < num)
            pq.offer(num);
        if(pq.size() == k+1)
            pq.poll();
    }
}

我使用了 Integer 的 PriorityQueue，但它很简单，可以用自定义对象替换它并输入自定义 Comparator。

除非我遗漏了一些明显的东西，否则我想这就是 OP 想要的。

【讨论】：

谢谢！这确实是一个简单的解决方案。但是，如果我的理解正确，它可能不是最理想的：真正的固定大小优先级队列会很快开始拒绝新提供的元素，因为它们中的大多数都低于队列中的元素。如果您还跟踪最低元素，则检查新元素将像比较一样容易。然而，在你的实现中，添加一个新元素总是会改变集合，这是很昂贵的。对您的实现的明显优化是将新元素与pq.peek() 进行比较（因为它是最低的），并且仅在它更大时才提供它。
@ffriend 是的，我已经编辑了答案以反映这一点。

【解决方案10】：

创建一个有大小限制的 PriorityQueue。它存储 N 个最大数字。

import java.util.*;

class Demo
{
    public static <E extends Comparable<E>> PriorityQueue<E> getPq(final int n, Comparator<E> comparator)
    {
        return new PriorityQueue<E>(comparator)
        {
            boolean full()
            {
                return size() >= n;
            }

            @Override 
            public boolean add(E e)
            {
                if (!full())
                {
                    return super.add(e);
                }
                else if (peek().compareTo(e) < 0)
                {
                    poll();
                    return super.add(e);
                }
                return false;
            }

            @Override
            public boolean offer(E e)
            {
                if (!full())
                {
                    return super.offer(e);
                }
                else if (peek().compareTo(e) < 0)
                {
                    poll();
                    return super.offer(e);
                }
                return false;
            }
        };
    }

    public static void printq(PriorityQueue pq)
    {
        Object o = null;
        while ((o = pq.poll()) != null)
        {
            System.out.println(o);
        }
    }

    public static void main (String[] args)
    {
        PriorityQueue<Integer> pq = getPq(2, new Comparator<Integer>(){
        @Override
        public int compare(Integer i1, Integer i2)
        {
            return i1.compareTo(i2);
        }
        });
        pq.add(4);
        pq.add(1);
        pq.add(5);
        pq.add(2);
        printq(pq);
    }
}

【讨论】：