2 * Written by Doug Lea with assistance from members of JCP JSR-166
3 * Expert Group and released to the public domain, as explained at
4 * http://creativecommons.org/licenses/publicdomain
7 package java.util.concurrent;
9 import java.util.concurrent.locks.*;
13 * An unbounded {@linkplain BlockingQueue blocking queue} that uses
14 * the same ordering rules as class {@link PriorityQueue} and supplies
15 * blocking retrieval operations. While this queue is logically
16 * unbounded, attempted additions may fail due to resource exhaustion
17 * (causing <tt>OutOfMemoryError</tt>). This class does not permit
18 * <tt>null</tt> elements. A priority queue relying on {@linkplain
19 * Comparable natural ordering} also does not permit insertion of
20 * non-comparable objects (doing so results in
21 * <tt>ClassCastException</tt>).
23 * <p>This class and its iterator implement all of the
24 * <em>optional</em> methods of the {@link Collection} and {@link
25 * Iterator} interfaces. The Iterator provided in method {@link
26 * #iterator()} is <em>not</em> guaranteed to traverse the elements of
27 * the PriorityBlockingQueue in any particular order. If you need
28 * ordered traversal, consider using
29 * <tt>Arrays.sort(pq.toArray())</tt>. Also, method <tt>drainTo</tt>
30 * can be used to <em>remove</em> some or all elements in priority
31 * order and place them in another collection.
33 * <p>Operations on this class make no guarantees about the ordering
34 * of elements with equal priority. If you need to enforce an
35 * ordering, you can define custom classes or comparators that use a
36 * secondary key to break ties in primary priority values. For
37 * example, here is a class that applies first-in-first-out
38 * tie-breaking to comparable elements. To use it, you would insert a
39 * <tt>new FIFOEntry(anEntry)</tt> instead of a plain entry object.
42 * class FIFOEntry<E extends Comparable<? super E>>
43 * implements Comparable<FIFOEntry<E>> {
44 * final static AtomicLong seq = new AtomicLong();
47 * public FIFOEntry(E entry) {
48 * seqNum = seq.getAndIncrement();
51 * public E getEntry() { return entry; }
52 * public int compareTo(FIFOEntry<E> other) {
53 * int res = entry.compareTo(other.entry);
54 * if (res == 0 && other.entry != this.entry)
55 * res = (seqNum < other.seqNum ? -1 : 1);
60 * <p>This class is a member of the
61 * <a href="{@docRoot}/../technotes/guides/collections/index.html">
62 * Java Collections Framework</a>.
66 * @param <E> the type of elements held in this collection
68 public class PriorityBlockingQueue<E> extends AbstractQueue<E>
69 implements BlockingQueue<E>, java.io.Serializable {
70 private static final long serialVersionUID = 5595510919245408276L;
72 private final PriorityQueue<E> q;
73 private final ReentrantLock lock = new ReentrantLock(true);
74 private final Condition notEmpty = lock.newCondition();
77 * Creates a <tt>PriorityBlockingQueue</tt> with the default
78 * initial capacity (11) that orders its elements according to
79 * their {@linkplain Comparable natural ordering}.
81 public PriorityBlockingQueue() {
82 q = new PriorityQueue<E>();
86 * Creates a <tt>PriorityBlockingQueue</tt> with the specified
87 * initial capacity that orders its elements according to their
88 * {@linkplain Comparable natural ordering}.
90 * @param initialCapacity the initial capacity for this priority queue
91 * @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
94 public PriorityBlockingQueue(int initialCapacity) {
95 q = new PriorityQueue<E>(initialCapacity, null);
99 * Creates a <tt>PriorityBlockingQueue</tt> with the specified initial
100 * capacity that orders its elements according to the specified
103 * @param initialCapacity the initial capacity for this priority queue
104 * @param comparator the comparator that will be used to order this
105 * priority queue. If {@code null}, the {@linkplain Comparable
106 * natural ordering} of the elements will be used.
107 * @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
110 public PriorityBlockingQueue(int initialCapacity,
111 Comparator<? super E> comparator) {
112 q = new PriorityQueue<E>(initialCapacity, comparator);
116 * Creates a <tt>PriorityBlockingQueue</tt> containing the elements
117 * in the specified collection. If the specified collection is a
118 * {@link SortedSet} or a {@link PriorityQueue}, this
119 * priority queue will be ordered according to the same ordering.
120 * Otherwise, this priority queue will be ordered according to the
121 * {@linkplain Comparable natural ordering} of its elements.
123 * @param c the collection whose elements are to be placed
124 * into this priority queue
125 * @throws ClassCastException if elements of the specified collection
126 * cannot be compared to one another according to the priority
128 * @throws NullPointerException if the specified collection or any
129 * of its elements are null
131 public PriorityBlockingQueue(Collection<? extends E> c) {
132 q = new PriorityQueue<E>(c);
136 * Inserts the specified element into this priority queue.
138 * @param e the element to add
139 * @return <tt>true</tt> (as specified by {@link Collection#add})
140 * @throws ClassCastException if the specified element cannot be compared
141 * with elements currently in the priority queue according to the
142 * priority queue's ordering
143 * @throws NullPointerException if the specified element is null
145 public boolean add(E e) {
150 * Inserts the specified element into this priority queue.
152 * @param e the element to add
153 * @return <tt>true</tt> (as specified by {@link Queue#offer})
154 * @throws ClassCastException if the specified element cannot be compared
155 * with elements currently in the priority queue according to the
156 * priority queue's ordering
157 * @throws NullPointerException if the specified element is null
159 public boolean offer(E e) {
160 final ReentrantLock lock = this.lock;
163 boolean ok = q.offer(e);
173 * Inserts the specified element into this priority queue. As the queue is
174 * unbounded this method will never block.
176 * @param e the element to add
177 * @throws ClassCastException if the specified element cannot be compared
178 * with elements currently in the priority queue according to the
179 * priority queue's ordering
180 * @throws NullPointerException if the specified element is null
182 public void put(E e) {
183 offer(e); // never need to block
187 * Inserts the specified element into this priority queue. As the queue is
188 * unbounded this method will never block.
190 * @param e the element to add
191 * @param timeout This parameter is ignored as the method never blocks
192 * @param unit This parameter is ignored as the method never blocks
193 * @return <tt>true</tt>
194 * @throws ClassCastException if the specified element cannot be compared
195 * with elements currently in the priority queue according to the
196 * priority queue's ordering
197 * @throws NullPointerException if the specified element is null
199 public boolean offer(E e, long timeout, TimeUnit unit) {
200 return offer(e); // never need to block
204 final ReentrantLock lock = this.lock;
213 public E take() throws InterruptedException {
214 final ReentrantLock lock = this.lock;
215 lock.lockInterruptibly();
218 while (q.size() == 0)
220 } catch (InterruptedException ie) {
221 notEmpty.signal(); // propagate to non-interrupted thread
232 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
233 long nanos = unit.toNanos(timeout);
234 final ReentrantLock lock = this.lock;
235 lock.lockInterruptibly();
244 nanos = notEmpty.awaitNanos(nanos);
245 } catch (InterruptedException ie) {
246 notEmpty.signal(); // propagate to non-interrupted thread
256 final ReentrantLock lock = this.lock;
266 * Returns the comparator used to order the elements in this queue,
267 * or <tt>null</tt> if this queue uses the {@linkplain Comparable
268 * natural ordering} of its elements.
270 * @return the comparator used to order the elements in this queue,
271 * or <tt>null</tt> if this queue uses the natural
272 * ordering of its elements
274 public Comparator<? super E> comparator() {
275 return q.comparator();
279 final ReentrantLock lock = this.lock;
289 * Always returns <tt>Integer.MAX_VALUE</tt> because
290 * a <tt>PriorityBlockingQueue</tt> is not capacity constrained.
291 * @return <tt>Integer.MAX_VALUE</tt>
293 public int remainingCapacity() {
294 return Integer.MAX_VALUE;
298 * Removes a single instance of the specified element from this queue,
299 * if it is present. More formally, removes an element {@code e} such
300 * that {@code o.equals(e)}, if this queue contains one or more such
301 * elements. Returns {@code true} if and only if this queue contained
302 * the specified element (or equivalently, if this queue changed as a
303 * result of the call).
305 * @param o element to be removed from this queue, if present
306 * @return <tt>true</tt> if this queue changed as a result of the call
308 public boolean remove(Object o) {
309 final ReentrantLock lock = this.lock;
319 * Returns {@code true} if this queue contains the specified element.
320 * More formally, returns {@code true} if and only if this queue contains
321 * at least one element {@code e} such that {@code o.equals(e)}.
323 * @param o object to be checked for containment in this queue
324 * @return <tt>true</tt> if this queue contains the specified element
326 public boolean contains(Object o) {
327 final ReentrantLock lock = this.lock;
330 return q.contains(o);
337 * Returns an array containing all of the elements in this queue.
338 * The returned array elements are in no particular order.
340 * <p>The returned array will be "safe" in that no references to it are
341 * maintained by this queue. (In other words, this method must allocate
342 * a new array). The caller is thus free to modify the returned array.
344 * <p>This method acts as bridge between array-based and collection-based
347 * @return an array containing all of the elements in this queue
349 public Object[] toArray() {
350 final ReentrantLock lock = this.lock;
360 public String toString() {
361 final ReentrantLock lock = this.lock;
371 * @throws UnsupportedOperationException {@inheritDoc}
372 * @throws ClassCastException {@inheritDoc}
373 * @throws NullPointerException {@inheritDoc}
374 * @throws IllegalArgumentException {@inheritDoc}
376 public int drainTo(Collection<? super E> c) {
378 throw new NullPointerException();
380 throw new IllegalArgumentException();
381 final ReentrantLock lock = this.lock;
386 while ( (e = q.poll()) != null) {
397 * @throws UnsupportedOperationException {@inheritDoc}
398 * @throws ClassCastException {@inheritDoc}
399 * @throws NullPointerException {@inheritDoc}
400 * @throws IllegalArgumentException {@inheritDoc}
402 public int drainTo(Collection<? super E> c, int maxElements) {
404 throw new NullPointerException();
406 throw new IllegalArgumentException();
407 if (maxElements <= 0)
409 final ReentrantLock lock = this.lock;
414 while (n < maxElements && (e = q.poll()) != null) {
425 * Atomically removes all of the elements from this queue.
426 * The queue will be empty after this call returns.
428 public void clear() {
429 final ReentrantLock lock = this.lock;
439 * Returns an array containing all of the elements in this queue; the
440 * runtime type of the returned array is that of the specified array.
441 * The returned array elements are in no particular order.
442 * If the queue fits in the specified array, it is returned therein.
443 * Otherwise, a new array is allocated with the runtime type of the
444 * specified array and the size of this queue.
446 * <p>If this queue fits in the specified array with room to spare
447 * (i.e., the array has more elements than this queue), the element in
448 * the array immediately following the end of the queue is set to
451 * <p>Like the {@link #toArray()} method, this method acts as bridge between
452 * array-based and collection-based APIs. Further, this method allows
453 * precise control over the runtime type of the output array, and may,
454 * under certain circumstances, be used to save allocation costs.
456 * <p>Suppose <tt>x</tt> is a queue known to contain only strings.
457 * The following code can be used to dump the queue into a newly
458 * allocated array of <tt>String</tt>:
461 * String[] y = x.toArray(new String[0]);</pre>
463 * Note that <tt>toArray(new Object[0])</tt> is identical in function to
464 * <tt>toArray()</tt>.
466 * @param a the array into which the elements of the queue are to
467 * be stored, if it is big enough; otherwise, a new array of the
468 * same runtime type is allocated for this purpose
469 * @return an array containing all of the elements in this queue
470 * @throws ArrayStoreException if the runtime type of the specified array
471 * is not a supertype of the runtime type of every element in
473 * @throws NullPointerException if the specified array is null
475 public <T> T[] toArray(T[] a) {
476 final ReentrantLock lock = this.lock;
486 * Returns an iterator over the elements in this queue. The
487 * iterator does not return the elements in any particular order.
488 * The returned <tt>Iterator</tt> is a "weakly consistent"
489 * iterator that will never throw {@link
490 * ConcurrentModificationException}, and guarantees to traverse
491 * elements as they existed upon construction of the iterator, and
492 * may (but is not guaranteed to) reflect any modifications
493 * subsequent to construction.
495 * @return an iterator over the elements in this queue
497 public Iterator<E> iterator() {
498 return new Itr(toArray());
502 * Snapshot iterator that works off copy of underlying q array.
504 private class Itr implements Iterator<E> {
505 final Object[] array; // Array of all elements
506 int cursor; // index of next element to return;
507 int lastRet; // index of last element, or -1 if no such
509 Itr(Object[] array) {
514 public boolean hasNext() {
515 return cursor < array.length;
519 if (cursor >= array.length)
520 throw new NoSuchElementException();
522 return (E)array[cursor++];
525 public void remove() {
527 throw new IllegalStateException();
528 Object x = array[lastRet];
530 // Traverse underlying queue to find == element,
531 // not just a .equals element.
534 for (Iterator it = q.iterator(); it.hasNext(); ) {
535 if (it.next() == x) {
547 * Saves the state to a stream (that is, serializes it). This
548 * merely wraps default serialization within lock. The
549 * serialization strategy for items is left to underlying
550 * Queue. Note that locking is not needed on deserialization, so
551 * readObject is not defined, just relying on default.
553 private void writeObject(java.io.ObjectOutputStream s)
554 throws java.io.IOException {
557 s.defaultWriteObject();