ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/PriorityBlockingQueue.java
Revision: 1.56
Committed: Mon Sep 20 20:23:52 2010 UTC (13 years, 8 months ago) by jsr166
Branch: MAIN
Changes since 1.55: +7 -7 lines
Log Message:
Fix javadoc code samples to use <pre>  {@code so that we can replace &lt; with <, etc.

File Contents

# Content
1 /*
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
5 */
6
7 package java.util.concurrent;
8
9 import java.util.concurrent.locks.*;
10 import java.util.*;
11
12 /**
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>).
22 *
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.
32 *
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.
40 *
41 * <pre> {@code
42 * class FIFOEntry<E extends Comparable<? super E>>
43 * implements Comparable<FIFOEntry<E>> {
44 * final static AtomicLong seq = new AtomicLong();
45 * final long seqNum;
46 * final E entry;
47 * public FIFOEntry(E entry) {
48 * seqNum = seq.getAndIncrement();
49 * this.entry = entry;
50 * }
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);
56 * return res;
57 * }
58 * }}</pre>
59 *
60 * <p>This class is a member of the
61 * <a href="{@docRoot}/../technotes/guides/collections/index.html">
62 * Java Collections Framework</a>.
63 *
64 * @since 1.5
65 * @author Doug Lea
66 * @param <E> the type of elements held in this collection
67 */
68 public class PriorityBlockingQueue<E> extends AbstractQueue<E>
69 implements BlockingQueue<E>, java.io.Serializable {
70 private static final long serialVersionUID = 5595510919245408276L;
71
72 final PriorityQueue<E> q;
73 final ReentrantLock lock = new ReentrantLock(true);
74 private final Condition notEmpty = lock.newCondition();
75
76 /**
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}.
80 */
81 public PriorityBlockingQueue() {
82 q = new PriorityQueue<E>();
83 }
84
85 /**
86 * Creates a <tt>PriorityBlockingQueue</tt> with the specified
87 * initial capacity that orders its elements according to their
88 * {@linkplain Comparable natural ordering}.
89 *
90 * @param initialCapacity the initial capacity for this priority queue
91 * @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
92 * than 1
93 */
94 public PriorityBlockingQueue(int initialCapacity) {
95 q = new PriorityQueue<E>(initialCapacity, null);
96 }
97
98 /**
99 * Creates a <tt>PriorityBlockingQueue</tt> with the specified initial
100 * capacity that orders its elements according to the specified
101 * comparator.
102 *
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
108 * than 1
109 */
110 public PriorityBlockingQueue(int initialCapacity,
111 Comparator<? super E> comparator) {
112 q = new PriorityQueue<E>(initialCapacity, comparator);
113 }
114
115 /**
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.
122 *
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
127 * queue's ordering
128 * @throws NullPointerException if the specified collection or any
129 * of its elements are null
130 */
131 public PriorityBlockingQueue(Collection<? extends E> c) {
132 q = new PriorityQueue<E>(c);
133 }
134
135 /**
136 * Inserts the specified element into this priority queue.
137 *
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
144 */
145 public boolean add(E e) {
146 return offer(e);
147 }
148
149 /**
150 * Inserts the specified element into this priority queue.
151 *
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
158 */
159 public boolean offer(E e) {
160 final ReentrantLock lock = this.lock;
161 lock.lock();
162 try {
163 boolean ok = q.offer(e);
164 assert ok;
165 notEmpty.signal();
166 return true;
167 } finally {
168 lock.unlock();
169 }
170 }
171
172 /**
173 * Inserts the specified element into this priority queue. As the queue is
174 * unbounded this method will never block.
175 *
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
181 */
182 public void put(E e) {
183 offer(e); // never need to block
184 }
185
186 /**
187 * Inserts the specified element into this priority queue. As the queue is
188 * unbounded this method will never block.
189 *
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
198 */
199 public boolean offer(E e, long timeout, TimeUnit unit) {
200 return offer(e); // never need to block
201 }
202
203 public E poll() {
204 final ReentrantLock lock = this.lock;
205 lock.lock();
206 try {
207 return q.poll();
208 } finally {
209 lock.unlock();
210 }
211 }
212
213 public E take() throws InterruptedException {
214 final ReentrantLock lock = this.lock;
215 lock.lockInterruptibly();
216 try {
217 E x;
218 while ( (x = q.poll()) == null)
219 notEmpty.await();
220 return x;
221 } finally {
222 lock.unlock();
223 }
224 }
225
226 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
227 long nanos = unit.toNanos(timeout);
228 final ReentrantLock lock = this.lock;
229 lock.lockInterruptibly();
230 try {
231 E x;
232 while ( (x = q.poll()) == null) {
233 if (nanos <= 0)
234 return null;
235 nanos = notEmpty.awaitNanos(nanos);
236 }
237 return x;
238 } finally {
239 lock.unlock();
240 }
241 }
242
243 public E peek() {
244 final ReentrantLock lock = this.lock;
245 lock.lock();
246 try {
247 return q.peek();
248 } finally {
249 lock.unlock();
250 }
251 }
252
253 /**
254 * Returns the comparator used to order the elements in this queue,
255 * or <tt>null</tt> if this queue uses the {@linkplain Comparable
256 * natural ordering} of its elements.
257 *
258 * @return the comparator used to order the elements in this queue,
259 * or <tt>null</tt> if this queue uses the natural
260 * ordering of its elements
261 */
262 public Comparator<? super E> comparator() {
263 return q.comparator();
264 }
265
266 public int size() {
267 final ReentrantLock lock = this.lock;
268 lock.lock();
269 try {
270 return q.size();
271 } finally {
272 lock.unlock();
273 }
274 }
275
276 /**
277 * Always returns <tt>Integer.MAX_VALUE</tt> because
278 * a <tt>PriorityBlockingQueue</tt> is not capacity constrained.
279 * @return <tt>Integer.MAX_VALUE</tt>
280 */
281 public int remainingCapacity() {
282 return Integer.MAX_VALUE;
283 }
284
285 /**
286 * Removes a single instance of the specified element from this queue,
287 * if it is present. More formally, removes an element {@code e} such
288 * that {@code o.equals(e)}, if this queue contains one or more such
289 * elements. Returns {@code true} if and only if this queue contained
290 * the specified element (or equivalently, if this queue changed as a
291 * result of the call).
292 *
293 * @param o element to be removed from this queue, if present
294 * @return <tt>true</tt> if this queue changed as a result of the call
295 */
296 public boolean remove(Object o) {
297 final ReentrantLock lock = this.lock;
298 lock.lock();
299 try {
300 return q.remove(o);
301 } finally {
302 lock.unlock();
303 }
304 }
305
306 /**
307 * Returns {@code true} if this queue contains the specified element.
308 * More formally, returns {@code true} if and only if this queue contains
309 * at least one element {@code e} such that {@code o.equals(e)}.
310 *
311 * @param o object to be checked for containment in this queue
312 * @return <tt>true</tt> if this queue contains the specified element
313 */
314 public boolean contains(Object o) {
315 final ReentrantLock lock = this.lock;
316 lock.lock();
317 try {
318 return q.contains(o);
319 } finally {
320 lock.unlock();
321 }
322 }
323
324 /**
325 * Returns an array containing all of the elements in this queue.
326 * The returned array elements are in no particular order.
327 *
328 * <p>The returned array will be "safe" in that no references to it are
329 * maintained by this queue. (In other words, this method must allocate
330 * a new array). The caller is thus free to modify the returned array.
331 *
332 * <p>This method acts as bridge between array-based and collection-based
333 * APIs.
334 *
335 * @return an array containing all of the elements in this queue
336 */
337 public Object[] toArray() {
338 final ReentrantLock lock = this.lock;
339 lock.lock();
340 try {
341 return q.toArray();
342 } finally {
343 lock.unlock();
344 }
345 }
346
347
348 public String toString() {
349 final ReentrantLock lock = this.lock;
350 lock.lock();
351 try {
352 return q.toString();
353 } finally {
354 lock.unlock();
355 }
356 }
357
358 /**
359 * @throws UnsupportedOperationException {@inheritDoc}
360 * @throws ClassCastException {@inheritDoc}
361 * @throws NullPointerException {@inheritDoc}
362 * @throws IllegalArgumentException {@inheritDoc}
363 */
364 public int drainTo(Collection<? super E> c) {
365 if (c == null)
366 throw new NullPointerException();
367 if (c == this)
368 throw new IllegalArgumentException();
369 final ReentrantLock lock = this.lock;
370 lock.lock();
371 try {
372 int n = 0;
373 E e;
374 while ( (e = q.poll()) != null) {
375 c.add(e);
376 ++n;
377 }
378 return n;
379 } finally {
380 lock.unlock();
381 }
382 }
383
384 /**
385 * @throws UnsupportedOperationException {@inheritDoc}
386 * @throws ClassCastException {@inheritDoc}
387 * @throws NullPointerException {@inheritDoc}
388 * @throws IllegalArgumentException {@inheritDoc}
389 */
390 public int drainTo(Collection<? super E> c, int maxElements) {
391 if (c == null)
392 throw new NullPointerException();
393 if (c == this)
394 throw new IllegalArgumentException();
395 if (maxElements <= 0)
396 return 0;
397 final ReentrantLock lock = this.lock;
398 lock.lock();
399 try {
400 int n = 0;
401 E e;
402 while (n < maxElements && (e = q.poll()) != null) {
403 c.add(e);
404 ++n;
405 }
406 return n;
407 } finally {
408 lock.unlock();
409 }
410 }
411
412 /**
413 * Atomically removes all of the elements from this queue.
414 * The queue will be empty after this call returns.
415 */
416 public void clear() {
417 final ReentrantLock lock = this.lock;
418 lock.lock();
419 try {
420 q.clear();
421 } finally {
422 lock.unlock();
423 }
424 }
425
426 /**
427 * Returns an array containing all of the elements in this queue; the
428 * runtime type of the returned array is that of the specified array.
429 * The returned array elements are in no particular order.
430 * If the queue fits in the specified array, it is returned therein.
431 * Otherwise, a new array is allocated with the runtime type of the
432 * specified array and the size of this queue.
433 *
434 * <p>If this queue fits in the specified array with room to spare
435 * (i.e., the array has more elements than this queue), the element in
436 * the array immediately following the end of the queue is set to
437 * <tt>null</tt>.
438 *
439 * <p>Like the {@link #toArray()} method, this method acts as bridge between
440 * array-based and collection-based APIs. Further, this method allows
441 * precise control over the runtime type of the output array, and may,
442 * under certain circumstances, be used to save allocation costs.
443 *
444 * <p>Suppose <tt>x</tt> is a queue known to contain only strings.
445 * The following code can be used to dump the queue into a newly
446 * allocated array of <tt>String</tt>:
447 *
448 * <pre>
449 * String[] y = x.toArray(new String[0]);</pre>
450 *
451 * Note that <tt>toArray(new Object[0])</tt> is identical in function to
452 * <tt>toArray()</tt>.
453 *
454 * @param a the array into which the elements of the queue are to
455 * be stored, if it is big enough; otherwise, a new array of the
456 * same runtime type is allocated for this purpose
457 * @return an array containing all of the elements in this queue
458 * @throws ArrayStoreException if the runtime type of the specified array
459 * is not a supertype of the runtime type of every element in
460 * this queue
461 * @throws NullPointerException if the specified array is null
462 */
463 public <T> T[] toArray(T[] a) {
464 final ReentrantLock lock = this.lock;
465 lock.lock();
466 try {
467 return q.toArray(a);
468 } finally {
469 lock.unlock();
470 }
471 }
472
473 /**
474 * Returns an iterator over the elements in this queue. The
475 * iterator does not return the elements in any particular order.
476 * The returned <tt>Iterator</tt> is a "weakly consistent"
477 * iterator that will never throw {@link
478 * ConcurrentModificationException}, and guarantees to traverse
479 * elements as they existed upon construction of the iterator, and
480 * may (but is not guaranteed to) reflect any modifications
481 * subsequent to construction.
482 *
483 * @return an iterator over the elements in this queue
484 */
485 public Iterator<E> iterator() {
486 return new Itr(toArray());
487 }
488
489 /**
490 * Snapshot iterator that works off copy of underlying q array.
491 */
492 private class Itr implements Iterator<E> {
493 final Object[] array; // Array of all elements
494 int cursor; // index of next element to return;
495 int lastRet; // index of last element, or -1 if no such
496
497 Itr(Object[] array) {
498 lastRet = -1;
499 this.array = array;
500 }
501
502 public boolean hasNext() {
503 return cursor < array.length;
504 }
505
506 public E next() {
507 if (cursor >= array.length)
508 throw new NoSuchElementException();
509 lastRet = cursor;
510 return (E)array[cursor++];
511 }
512
513 public void remove() {
514 if (lastRet < 0)
515 throw new IllegalStateException();
516 Object x = array[lastRet];
517 lastRet = -1;
518 // Traverse underlying queue to find == element,
519 // not just a .equals element.
520 lock.lock();
521 try {
522 for (Iterator it = q.iterator(); it.hasNext(); ) {
523 if (it.next() == x) {
524 it.remove();
525 return;
526 }
527 }
528 } finally {
529 lock.unlock();
530 }
531 }
532 }
533
534 /**
535 * Saves the state to a stream (that is, serializes it). This
536 * merely wraps default serialization within lock. The
537 * serialization strategy for items is left to underlying
538 * Queue. Note that locking is not needed on deserialization, so
539 * readObject is not defined, just relying on default.
540 */
541 private void writeObject(java.io.ObjectOutputStream s)
542 throws java.io.IOException {
543 lock.lock();
544 try {
545 s.defaultWriteObject();
546 } finally {
547 lock.unlock();
548 }
549 }
550
551 }