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

Question

相关问题：

• 具有固定大小的Java PriorityQueue
• 如何使用 PriorityQueue？
• 获取数组中 n 个最小元素的索引
• Scala：有没有办法使用PriorityQueue 就像我在 Java 中那样？

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

• scala.collection.mutable.PriorityQueue
• java.util.PriorityQueue
• lucene.util.PriorityQueue

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

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

UPD。最后，我根据Peter的回答创建了自己的实现：（

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取自这个问题）

Answer 1

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

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

Answer 2

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

Answer 3

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

Answer 4

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

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

Answer 5

下面是我之前使用的实现。符合Peter的建议。

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

Answer 6

正是我正在寻找的东西。然而，该实现包含一个错误：

即：if 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;

Answer 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;

    }
}

}

Answer 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);
    }
}

Answer 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，但它很简单，可以用自定义对象替换它并输入自定义比较器。

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

Answer 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);
    }
}