java队列--Queue(一)

╰半橙微兮° 2022-03-19 16:09 146阅读 0赞

### Java队列Queue之第一篇 ###

*  定义
 *  类图
 *  队列的实现
 *   *  非阻塞队列
     *   *  PriorityQueue
         *  ConcurrentLinkedQueue
     *  阻塞队列
     *   *  ArrayBlockingQueue
         *   *  基本属性
         *  LinkedBlockingQueue
         *   *  基本属性
         *  PriorityBlockingQueue
         *   *  基本属性
         *  DelayQueue
         *   *  基本属性
 *  公用方法

# 定义 #

队列，又称为伫列（`queue`），是先进先出（`FIFO`, `First-In-First-Out`）的线性表。在具体应用中通常用链表或者数组来实现。队列只允许在后端（称为`rear`）进行插入操作，在前端（称为`front`）进行删除操作。队列的操作方式和堆栈类似，唯一的区别在于队列只允许新数据在后端进行添加。

# 类图 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTYxMDU1_size_16_color_FFFFFF_t_70]  
Queue接口与List、Set同一级别，都是继承了Collection接口。LinkedList实现了Deque接 口。

# 队列的实现 #

## 非阻塞队列 ##

非阻塞队列实现了`java.util.Queue`接口和`java.util.AbstractQueue`接口,在JDK中的类也比较少，包括`PriorityQueue`和`ConcurrentLinkedQueue`两种。

### PriorityQueue ###

`PriorityQueue` 维护了一个有序列表。加入到 `Queue` 中的元素根据它们的天然排序（通过其 `java.util.Comparable` 实现）或者根据传递给构造函数的 `java.util.Comparator` 实现来定位。  
下面看下具体的JDK源码：

/**
    	 * 优先级队列表示为平衡二进制堆，其优先级是依据比较器的，如果没有比较器，则按照节点的自然顺序。
         * Priority queue represented as a balanced binary heap: the two
         * children of queue[n] are queue[2*n+1] and queue[2*(n+1)].  The
         * priority queue is ordered by comparator, or by the elements'
         * natural ordering, if comparator is null: For each node n in the
         * heap and each descendant d of n, n <= d.  The element with the
         * lowest value is in queue[0], assuming the queue is nonempty.
         */
        transient Object[] queue; // non-private to simplify nested class access
        
    	/**
    	 * 队列中的元素个数(并不是上面的数组的长度，数组长度可以在初始化时指定，但未必放满元素，
    	 * 即元素个数==数组长度不是恒等的)
         * The number of elements in the priority queue.
         */
        private int size = 0;
    
        /**
         * 比较器
         * The comparator, or null if priority queue uses elements'
         * natural ordering.
         */
        private final Comparator<? super E> comparator;
    
        /**
         * 队列结构修改的次数
         * The number of times this priority queue has been
         * <i>structurally modified</i>.  See AbstractList for gory details.
         */
        transient int modCount = 0; // non-private to simplify nested class access

其它的就是一些构造器(数组初始化长度或者比较器作为参数)和数组的一些操作的方法，还有一些通用的方法，后面会讲解到。

### ConcurrentLinkedQueue ###

`ConcurrentLinkedQueue` 则是基于链表的、线程安全的队列。

/**
      * 维护的静态内部类
      */
    private static class Node<E> {
            volatile E item;
            volatile Node<E> next;
    
            /**
             * Constructs a new node.  Uses relaxed write because item can
             * only be seen after publication via casNext.
             */
            Node(E item) {
                UNSAFE.putObject(this, itemOffset, item);
            }
    
            boolean casItem(E cmp, E val) {
                return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
            }
    
            void lazySetNext(Node<E> val) {
                UNSAFE.putOrderedObject(this, nextOffset, val);
            }
    
            boolean casNext(Node<E> cmp, Node<E> val) {
                return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
            }
    
            // Unsafe mechanics
    
            private static final sun.misc.Unsafe UNSAFE;
            private static final long itemOffset;
            private static final long nextOffset;
    
            static {
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class<?> k = Node.class;
                    itemOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("item"));
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }
        }

在头节点和尾部节点都加了`volatile`关键字修饰。因为它在队列的尾部添加元素并从头部删除它们，所以只要不需要知道队列的大 小，`ConcurrentLinkedQueue` 对公共集合的共享访问就可以工作得很好。获取队列的长度效率比较低，因为要遍历链表。

## 阻塞队列 ##

阻塞队列都是实现了`BlockingQueue`接口的，包括五个实现类，分别是：`ArrayBlockingQueue`、`LinkedBlockingQueue`、`PriorityBlockingQueue`、`DelayQueue`、`SynchronousQueue`。

### ArrayBlockingQueue ###

在构造时需要指定容量， 并可以选择是否需要公平性，如果公平参数被设置`true`，等待时间最长的线程会优先得到处理（其实就是通过将`ReentrantLock`设置为`true`来 达到这种公平性的：即等待时间最长的线程会先操作）。通常，公平性会使你在性能上付出代价，只有在的确非常需要的时候再使用它。它是基于数组的阻塞循环队 列，此队列按 `FIFO`（先进先出）原则对元素进行排序。

#### 基本属性 ####

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
    	final Object[] items; //用来存储元素的数组
    	// 下一次执行task,poll,peek或者remove的索引
    	 int takeIndex;
    	// 下一次执行put,offer,或者add的索引
    	 int putIndex;
    	//当前数组中存放的元素数量
    	 int count;
    	// 提供的lock锁
    	 final ReentrantLock lock;
    	// 队列空了，就挂起，等到有元素放进去后唤醒
    	 private final Condition notEmpty;
    	// 队列满了就挂起，等到取出一个元素后唤醒
    	 private final Condition notFull;
    }

由此可见，也是基于数组的。切引入了重入锁和两个 `Condition` 。且是有界的。

### LinkedBlockingQueue ###

基于链表的有界队列。在不指定容量时存放元素个数没有限制。

#### 基本属性 ####

/** The capacity bound, or Integer.MAX_VALUE if none */
        private final int capacity;
    
        /** Current number of elements */
        private final AtomicInteger count = new AtomicInteger();
    
        /**
         * Head of linked list.
         * Invariant: head.item == null
         */
        transient Node<E> head;
    
        /**
         * Tail of linked list.
         * Invariant: last.next == null
         */
        private transient Node<E> last;
    
        /** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock();
    
        /** Wait queue for waiting takes */
        private final Condition notEmpty = takeLock.newCondition();
    
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock();
    
        /** Wait queue for waiting puts */
        private final Condition notFull = putLock.newCondition();

### PriorityBlockingQueue ###

是一个带优先级的 队列，而不是先进先出队列。元素按优先级顺序被移除，该队列也没有上限（`PriorityBlockingQueue`是对 `PriorityQueue`的再次包装，是基于堆数据结构的，而`PriorityQueue`是没有容量限制的，与`ArrayList`一样，所以在优先阻塞 队列上`put`时是不会受阻的。虽然此队列逻辑上是无界的，但是由于资源被耗尽，所以试图执行添加操作可能会导致 `OutOfMemoryError`），但是如果队列为空，那么取元素的操作`take`就会阻塞，所以它的检索操作`take`是受阻的。另外，往入该队列中的元 素要具有比较能力。

#### 基本属性 ####

/**
         * Default array capacity.
         */
        private static final int DEFAULT_INITIAL_CAPACITY = 11;
    
        /**
         * The maximum size of array to allocate.
         * Some VMs reserve some header words in an array.
         * Attempts to allocate larger arrays may result in
         * OutOfMemoryError: Requested array size exceeds VM limit
         */
        private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    
        /**
         * Priority queue represented as a balanced binary heap: the two
         * children of queue[n] are queue[2*n+1] and queue[2*(n+1)].  The
         * priority queue is ordered by comparator, or by the elements'
         * natural ordering, if comparator is null: For each node n in the
         * heap and each descendant d of n, n <= d.  The element with the
         * lowest value is in queue[0], assuming the queue is nonempty.
         */
        private transient Object[] queue;
    
        /**
         * The number of elements in the priority queue.
         */
        private transient int size;
    
        /**
         * The comparator, or null if priority queue uses elements'
         * natural ordering.
         */
        private transient Comparator<? super E> comparator;
    
        /**
         * Lock used for all public operations
         */
        private final ReentrantLock lock;
    
        /**
         * Condition for blocking when empty
         */
        private final Condition notEmpty;
    
        /**
         * Spinlock for allocation, acquired via CAS.
         */
        private transient volatile int allocationSpinLock;
    
        /**
         * A plain PriorityQueue used only for serialization,
         * to maintain compatibility with previous versions
         * of this class. Non-null only during serialization/deserialization.
         */
        private PriorityQueue<E> q;

### DelayQueue ###

（基于`PriorityQueue`来实现的）是一个存放`Delayed` 元素的无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 `Delayed` 元素。如果延迟都还没有期满，则队列没有头部，并且`poll`将返回`null`。当一个元素的 `getDelay(TimeUnit.NANOSECONDS)`方法返回一个小于或等于零的值时，则出现期满，`poll` 就以移除这个元素了。此队列不允许使用 `null` 元素。

#### 基本属性 ####

private final transient ReentrantLock lock = new ReentrantLock();
        private final PriorityQueue<E> q = new PriorityQueue<E>();
    
        /**
         * Thread designated to wait for the element at the head of
         * the queue.  This variant of the Leader-Follower pattern
         * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
         * minimize unnecessary timed waiting.  When a thread becomes
         * the leader, it waits only for the next delay to elapse, but
         * other threads await indefinitely.  The leader thread must
         * signal some other thread before returning from take() or
         * poll(...), unless some other thread becomes leader in the
         * interim.  Whenever the head of the queue is replaced with
         * an element with an earlier expiration time, the leader
         * field is invalidated by being reset to null, and some
         * waiting thread, but not necessarily the current leader, is
         * signalled.  So waiting threads must be prepared to acquire
         * and lose leadership while waiting.
         */
        private Thread leader = null;
    
        /**
         * Condition signalled when a newer element becomes available
         * at the head of the queue or a new thread may need to
         * become leader.
         */
        private final Condition available = lock.newCondition();

# 公用方法 #

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>作用</th> 
   <th>异常情况</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>add</code></td> 
   <td>增加一个元索</td> 
   <td>如果队列已满，则抛出一个<code>IIIegaISlabEepeplian</code>异常</td> 
  </tr> 
  <tr> 
   <td><code>remove</code></td> 
   <td>移除并返回队列头部的元素</td> 
   <td>如果队列为空，则抛出一个<code>NoSuchElementException</code>异常</td> 
  </tr> 
  <tr> 
   <td><code>element</code></td> 
   <td>返回队列头部的元素</td> 
   <td>如果队列为空，则抛出一个<code>NoSuchElementException</code>异常</td> 
  </tr> 
  <tr> 
   <td><code>offer</code></td> 
   <td>添加一个元素并返回<code>true</code></td> 
   <td>如果队列已满，则返回<code>false</code></td> 
  </tr> 
  <tr> 
   <td><code>poll</code></td> 
   <td>移除并返问队列头部的元素</td> 
   <td>如果队列为空，则返回<code>null</code></td> 
  </tr> 
  <tr> 
   <td><code>peek</code></td> 
   <td>返回队列头部的元素</td> 
   <td>如果队列为空，则返回<code>null</code></td> 
  </tr> 
  <tr> 
   <td><code>put</code></td> 
   <td>添加一个元素</td> 
   <td>如果队列满，则阻塞</td> 
  </tr> 
  <tr> 
   <td><code>take</code></td> 
   <td>移除并返回队列头部的元素</td> 
   <td>如果队列为空，则阻塞</td> 
  </tr> 
 </tbody> 
</table>

> 目前的说明比较浅显，对各个队列中的核心方法没有进一步说明和分析。在后面的章节中会继续逐一分析并对每种队列的应用场景和使用方法做详细分析。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTYxMDU1_size_16_color_FFFFFF_t_70]: /images/20220319/c76f28973c5a4356ab50c6a7d36b2b0e.png