Java之Queue

阳光穿透心脏的1/2处 2023-06-11 05:27 25阅读 0赞

本文链接： [https://blog.csdn.net/bingzhixi/article/details/78701036][https_blog.csdn.net_bingzhixi_article_details_78701036]

Queue是一种很常见的[数据结构][Link 2]类型，在[Java][]里面Queue是一个接口，它只是定义了一个基本的Queue应该有哪些功能规约。实际上有多个Queue的实现，有的是采用线性表实现，有的基于链表实现。还有的适用于多线程的环境。java中具有Queue功能的类主要有如下几个：AbstractQueue, ArrayBlockingQueue, ConcurrentLinkedQueue, LinkedBlockingQueue, DelayQueue, LinkedList, PriorityBlockingQueue, PriorityQueue和ArrayDqueue。

# Queue #

Queue本身是一种先入先出的模型(FIFO)，和我们日常生活中的排队模型很类似。根据不同的实现，他们主要有数组和链表两种实现形式。如下图：

![Center][]

![Center 1][]

因为在队列里和我们日常的模型很近似，每次如果要出队的话，都是从队头移除。而如果每次要加入新的元素，则要在队尾加。所以我们要在队列里保存队头和队尾。

在jdk里几个常用队列实现之间的类关系图如下：

![Center 2][]

可以看到，Deque也是一个接口，它继承了Queue的接口规范。Queue作为一个接口，它声明的几个基本操作无非就是入队和出队的操作，具体定义如下：

public interface Queue<E> extends Collection<E> { 
       
       boolean add(E e); // 添加元素到队列中，相当于进入队尾排队。 
       
       boolean offer(E e); //添加元素到队列中，相当于进入队尾排队. 
       
       E remove(); //移除队头元素 
       
       E poll(); //移除队头元素 
       
       E element(); //获取但不移除队列头的元素 
       
       E peek(); //获取但不移除队列头的元素 
       }

有了这些接口定义的规约，我们就可以很容易的在后续的详细实现里察看具体细节。

# Deque #

按照我们一般的理解，Deque是一个双向队列，这将意味着它不过是对Queue接口的增强。

如果仔细分析Deque接口代码的话，我们会发现它里面主要包含有4个部分的功能定义。

1. 双向队列特定方法定义。

2. Queue方法定义。

3. Stack方法定义。

4. Collection方法定义。

第3，4部分的方法相当于告诉我们，具体实现Deque的类我们也可以把他们当成Stack和普通的Collection来使用。这也是接口定义规约带来的好处。这里我们就不再赘述。DeQueue的定义如下：

public interface Deque<E> extends Queue<E> { 
       
       void addFirst(E e); 
       
       void addLast(E e); 
       
       boolean offerFirst(E e); 
       
       boolean offerLast(E e); 
       
       E removeFirst(); 
       
       E removeLast(); 
       
       E pollFirst(); 
       
       E pollLast(); 
       
       E getFirst(); 
       
       E getLast(); 
       
       E peekFirst(); 
       
       E peekLast(); 
       
       boolean removeFirstOccurrence(Object o); 
       
       boolean removeLastOccurrence(Object o); 
       
       // *** Queue methods *** 
       boolean add(E e); 
       
       boolean offer(E e); 
       
       E remove(); 
       
       E poll(); 
       
       E element(); 
       
       E peek(); 
       
       // *** Stack methods *** 
       void push(E e); 
       
       E pop(); 
       
       // *** Collection methods *** 
       boolean remove(Object o); 
       
       boolean contains(Object o); 
       
       public int size(); 
       
       Iterator<E> iterator(); 
       
       Iterator<E> descendingIterator(); 
       }

下面将分别简单介绍几种常用的队列：ArrayDeque、ArrayBlockingQueue、PriorityQuue、PriorityBlockingQueue、LinkedBlockingDeque、LinkedBlockingQueue、ConcurrentLinkedDeque、ConcurrentLinkedQueue

**1.ArrayDeque**

ArrayDeque继承自AbstractCollection，是最简单的队列，有以下特征：

a. 采用数组存储数据，

b. 扩容也是采用2倍扩容的方式

c. 有head和tail两个成员变量，在队尾插入数据时，tail加1，从队头弹出数据时head+1，当head==tail时，便进行扩容。

ArrayDeque中在队尾插入数据的源码如下：

public void addLast(E e) { 
       if (e == null) 
       throw new NullPointerException(); 
       elements[tail] = e; 
       if ( (tail = (tail + 1) & (elements.length - 1)) == head) 
       doubleCapacity(); 
       }

ArrayDeque的实现方式很简单，在此就不做详细的介绍了。

**2.ArrayBlockingQueue**

ArrayBlockingQueue也是采用数组存储数据的，但是它继承自AbstractQueue，之所以它是Blocking的是因为当put或poll数据时，如果Queue是满的或者没有数据，当前线程将阻塞等待，直到满足条件或当前线程被其他线程中断。它通过ReetranLock及其两个信号量notfFull和notEmpty实现线程安全的。构造函数如下：

public ArrayBlockingQueue(int capacity, boolean fair) { 
       if (capacity <= 0) 
       throw new IllegalArgumentException(); 
       this.items = new Object[capacity]; 
       lock = new ReentrantLock(fair); 
       notEmpty = lock.newCondition(); 
       notFull = lock.newCondition(); 
       }

ArrayBlockingQueue表示队首和队尾的成员变量分别是tableIndex和putIndex，定义如下：

/** items index for next take, poll, peek or remove */ 
       int takeIndex; 
       
       /** items index for next put, offer, or add */ 
       int putIndex;

插入数据时，先获取lock锁和notFull信号量，之后再插入数据，插入完成后释放lock锁并重置notEmpty信号量，源码如下：

public void put(E e) throws InterruptedException { 
       checkNotNull(e); 
       final ReentrantLock lock = this.lock; 
       lock.lockInterruptibly(); 
       try { 
       while (count == items.length) 
       notFull.await(); 
       enqueue(e); 
       } finally { 
       lock.unlock(); 
       } 
       } 
       private void enqueue(E x) { 
       // assert lock.getHoldCount() == 1; 
       // assert items[putIndex] == null; 
       final Object[] items = this.items; 
       items[putIndex] = x; 
       if (++putIndex == items.length) 
       putIndex = 0; 
       count++; 
       notEmpty.signal(); 
       }

弹出数据时，也是要先获取到lock锁的，弹出后需要重置notFull信号量，源码如下：

public E poll() { 
       final ReentrantLock lock = this.lock; 
       lock.lock(); 
       try { 
       return (count == 0) ? null : dequeue(); 
       } finally { 
       lock.unlock(); 
       } 
       } 
       private E dequeue() { 
       // assert lock.getHoldCount() == 1; 
       // assert items[takeIndex] != null; 
       final Object[] items = this.items; 
       @SuppressWarnings( "unchecked") 
       E x = (E) items[takeIndex]; 
       items[takeIndex] = null; 
       if (++takeIndex == items.length) 
       takeIndex = 0; 
       count--; 
       if (itrs != null) 
       itrs.elementDequeued(); 
       notFull.signal(); 
       return x; 
       }

除了通用的put和poll方法外，ArrayBlockingQueue还提供了带定时参数的put和poll方法，即如果无法取得相应的信号量，将等待固定的时间后返回，源码如下：

public boolean offer(E e, long timeout, TimeUnit unit) 
       throws InterruptedException { 
       
       checkNotNull(e); 
       long nanos = unit.toNanos(timeout); 
       final ReentrantLock lock = this.lock; 
       lock.lockInterruptibly(); 
       try { 
       while (count == items.length) { 
       if (nanos <= 0) 
       return false; 
       nanos = notFull.awaitNanos(nanos); 
       } 
       enqueue(e); 
       return true; 
       } finally { 
       lock.unlock(); 
       } 
       } 
       
       public E poll(long timeout, TimeUnit unit) throws InterruptedException { 
       long nanos = unit.toNanos(timeout); 
       final ReentrantLock lock = this.lock; 
       lock.lockInterruptibly(); 
       try { 
       while (count == 0) { 
       if (nanos <= 0) 
       return null; 
       nanos = notEmpty.awaitNanos(nanos); 
       } 
       return dequeue(); 
       } finally { 
       lock.unlock(); 
       } 
       }

总结一下ArrayBlockingQueue的特征有以下几点：

a. 用数组存储数据

b. 容量固定，不支持扩容，构造时必须插入容量值

c. 通过同步锁和信号量实现了线程安全。

d. 实现了Queue接口的方法，同时提供超时返回的offer和poll方法

e. 由于是线程安全的，因此不会有fail-fast问题。

**3.PriorityQueue**

PriorityQuue继承自AbstractQueue，通过构造函数传入的Comparator或者数据成员实现的Comparable接口的方法来确定成员间的优先级。PriorityQueue也是采用数组存储数据的，它只保证队首是最小的，其他的不保证顺序。插入时，会采用二分法将数据插入到最前面（注意，PriorityQueue是支持扩容的）插入数据的源码如下：

public boolean offer(E e) { 
       if (e == null) 
       throw new NullPointerException(); 
       modCount++; 
       int i = size; 
       if (i >= queue.length) 
       grow(i + 1); 
       size = i + 1; 
       if (i == 0) 
       queue[ 0] = e; 
       else 
       siftUp(i, e); 
       return true; 
       } 
       
       private void siftUp(int k, E x) { 
       if (comparator != null) 
       siftUpUsingComparator(k, x); 
       else 
       siftUpComparable(k, x); 
       } 
       
       @SuppressWarnings( "unchecked") 
       private void siftUpComparable(int k, E x) { 
       Comparable<? super E> key = (Comparable<? super E>) x; 
       while (k > 0) { 
       int parent = (k - 1) >>> 1; 
       Object e = queue[parent]; 
       if (key.compareTo((E) e) >= 0) 
       break; 
       queue[k] = e; 
       k = parent; 
       } 
       queue[k] = key; 
       } 
       
       @SuppressWarnings( "unchecked") 
       private void siftUpUsingComparator(int k, E x) { 
       while (k > 0) { 
       int parent = (k - 1) >>> 1; 
       Object e = queue[parent]; 
       if (comparator.compare(x, (E) e) >= 0) 
       break; 
       queue[k] = e; 
       k = parent; 
       } 
       queue[k] = x; 
       }

从源码分析，节点 k 的父节点的索引为 k>>>1(即k/2），父节点的值小于子节点。

弹出队首数据时，先将队首数据保存，然后将队尾数据放到队首后再按照父节点小于子节点的原则下降，弹出数据源码如下：

public E poll() { 
       if (size == 0) 
       return null; 
       int s = --size; 
       modCount++; 
       E result = (E) queue[ 0]; 
       E x = (E) queue[s]; 
       queue[s] = null; 
       if (s != 0) 
       siftDown( 0, x); 
       return result; 
       } 
       
       private void siftDown(int k, E x) { 
       if (comparator != null) 
       siftDownUsingComparator(k, x); 
       else 
       siftDownComparable(k, x); 
       } 
       
       @SuppressWarnings( "unchecked") 
       private void siftDownComparable(int k, E x) { 
       Comparable<? super E> key = (Comparable<? super E>)x; 
       int half = size >>> 1; // loop while a non-leaf 
       while (k < half) { 
       int child = (k << 1) + 1; // assume left child is least 
       Object c = queue[child]; 
       int right = child + 1; 
       if (right < size && 
       ((Comparable<? super E>) c).compareTo((E) queue[right]) > 0) 
       c = queue[child = right]; 
       if (key.compareTo((E) c) <= 0) 
       break; 
       queue[k] = c; 
       k = child; 
       } 
       queue[k] = key; 
       } 
       
       @SuppressWarnings( "unchecked") 
       private void siftDownUsingComparator(int k, E x) { 
       int half = size >>> 1; 
       while (k < half) { 
       int child = (k << 1) + 1; 
       Object c = queue[child]; 
       int right = child + 1; 
       if (right < size && 
       comparator.compare((E) c, (E) queue[right]) > 0) 
       c = queue[child = right]; 
       if (comparator.compare(x, (E) c) <= 0) 
       break; 
       queue[k] = c; 
       k = child; 
       } 
       queue[k] = x; 
       }

总结一下PriorityQueue的特征如下：

a. 采用数组存储数据

b. 支持扩容，当长度小于64时按+2扩容，否则按2倍扩容

c. 按Comparable接口或Comparator确定优先级

d. 队首的数据是最小的，但不保证其他数据的顺序。

e. 节点k的父节点是k/2，父节点的值小于子节点。

**4.PriorityBlockingQueue**

PriorityBlockingQueue继承自AbstractQueue，实现同PriorityQueue大同小异，只不过在对数据进行操作前使用了ReentrantLock实现了线程同步，PriorityBlockingQueue具有以下特征：

a. 具有所有PriorityQueue的特征

b. 采用同步锁实现线程安全

c. 由于支持扩容，因此add数据时，只需获取lock锁，不需要获取信号量

d. poll时需要获取notEmpty信号量，同时提供了超时退出的poll方法

**5. LinkedBlockingDeque**

LinkedBlockingDeque继承自AbstractQueue，采用链表结构存储数据，同时使用ReentrantLock保证线程安全，同时有notEmpty和notFull两个信号量。总结如下：

a. 采用链表存储数据，但容量固定，默认容量为最大整数。

b. 使用同步锁保证线程安全

c. 可以在队首和队尾插入或删除数据

d. 插入数据时，先获取lock，然后获取notFull信号量，插入后重置notEmpty信号量

e. 弹出数据时，先获取lock，得到数据后，重置notFull信号量（无数据时pop和poll方法会返回，takeFirst和takeLast会等待notEmpty信号量）

f. 提供了插入或弹出数据对应的超时返回方法。

**6. LinedBlockingQueue**

LinkedBlockingQueue也是继承自AbstractQueue，采用链表存储数据，它和LinkedBlockingDeque的区别在于，它有两把同步锁putLock和takeLock，分别控制数据的写入和弹出，并且它只能在队尾插入数据，从队首弹出数据。总结如下：

a. 采用链表存储数据，但容量固定，默认容量为最大整数

b. 读写分离的，有两把同步锁：写锁putLock和读锁takeLock

c. 只能在队尾插入数据，从队首弹出数据

d. 插入数据时，获取putLock锁，并等待notFull信号量，插入数据完成后，重置notEmpty信号量，如果队列没满则重置notFull信号量。

e. 弹出数据时，获取takeLock锁，获取到数据后，重置notFull信号量，如果队列仍不为空，重置notEmpty信号量。

**7.ConcurrentLinkedQueue**

ConcurrentLinkedQueue继承自AbstractQueue，采用链表存储数据，与上述几个实现线程的Queue方法不同，它通过Unsafe的CAS（compareAndSwap）方法实现线程安全（Unsafe相当于在JVM底层通过操作内存修改数据，如果在修改过程中数据没改变则替换成要修改的值，否则修改失败），总结如下：

a. 用链表存储数据，没有限制容量

b. 使用Unsafe类实现线程安全

[Link 1]: http://creativecommons.org/licenses/by-sa/4.0/
[https_blog.csdn.net_bingzhixi_article_details_78701036]: https://blog.csdn.net/bingzhixi/article/details/78701036
[Link 2]: http://lib.csdn.net/base/31
[Java]: http://lib.csdn.net/base/17
[Center]: /images/20230531/a46817def52f49bfa14254766112f80d.png
[Center 1]: /images/20230531/a329d3d681674ac78a04fbeae60680d6.png
[Center 2]: /images/20230531/cbc8caa85a1345028d1ab5cc5b7c36fb.png