Queue

Queue

  可装载元素的容器，提供了添加元素，移出队列头元素，检查队列头部是否有元素等方法。

 

 

	抛出异常	返回特殊值
插入	add(e)	offer(e)
移除	remove()	poll()
检查	element()	peek()

 

 

Queue 子类：

  

 

 Queue  方法：

 

boolean	add(E e)           将指定的元素插入此队列（如果立即可行且不会违反容量限制），在成功时返回 true，如果当前没有可用的空间，则抛出 IllegalStateException。
boolean	offer(E e)           将指定的元素插入此队列（如果立即可行且不会违反容量限制），则返回 true；否则返回 false
E	element()            获取，但是不移除此队列的头。如果此队列为空 抛出 NoSuchElementException -
E	peek()           获取但不移除此队列的头；如果此队列为空，则返回 null。
E	poll()           获取并移除此队列的头，如果此队列为空，则返回 null。
E	remove()           获取并移除此队列的头。如果此队列为空 抛出 NoSuchElementException

 

 

 

Queue 的选择：

 

 

 

单线程下：

 

        被添加的元素可以自动排序的无界队列（从头部到尾部按从小到大排序），就选择PriorityQueue

 

        需要双端插入移除无界队列，或后进先出无界队列或先进后出无界队列，可以选择 ArrayDeque

 

多线程下：

 

       不需要阻塞功能，先进先出的元界队列，就选择ConcurrentLinkedQueue（自旋锁）

 

       需要阻塞功能，先进先出的有界队列，线程少，就选择 ArrayBlockingQueue（可重入锁）（插入移除操作速度快）

 

 

       需要阻塞功能，先进先出的无界队列，线程多，就选择  LinkedBlockingQueue（读取写入各一把可重入锁），吞吐量大。

 

 

 

       需要阻塞功能，双端插入移除无界队列，或后进先出无界队列，就可以使LinkedBlockingDeque

 

      需要阻塞功能，需要延迟功能的无界队列，就选择DelayQueue

 

       需要阻塞功能，被添加的元素可以自动排序的无界队列（从头部到尾部按从小到大排序）,就可以使用PriorityBlockingQueue

 

       需要阻塞功能，0容量，仅为了把一个元素交给另一个线程，（另一线程接收前一直阻塞）使用SynchronousQueue

 

     

 

 

 

ConcurrentLinkedQueue:

 内部采用链表实现的无界队列。（不会出现违反容量限制的情况）

先进先出队列，新增的元素插入到队列尾部，从头部获取元素。

 

 线程安全的队列，实现线程安全的方式：采用CAS(CompareAndSwap) 方式，不断尝试直至成功。

源码：

 

private static final
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
        tailUpdater =
        AtomicReferenceFieldUpdater.newUpdater
        (ConcurrentLinkedQueue.class, Node.class, "tail");

//通过反射方式，如果对象的属性值与期望值相等，则原子的为属性赋新值。

    private boolean casTail(Node<E> cmp, Node<E> val) {
        return tailUpdater.compareAndSet(this, cmp, val);
    }
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        Node<E> n = new Node<E>(e, null);
        for (;;) {//不断循环
            Node<E> t = tail;
            Node<E> s = t.getNext();
            if (t == tail) {
                if (s == null) {
                    if (t.casNext(s, n)) {
//如果tail.next=s?则原子的设定tail.next=n，设定成功返回true,否则返回false
//false 说明有别的线程插入元素了，下一个循环再试插入元素
                        casTail(t, n);
                        return true;
                    }
                } else {
                    casTail(t, s);
                }
            }
        }
    }

 

 PriorityQueue:（优先级队列）

 内部采用数组来实现的有序无界队列。

排序方式：队列头部至尾部 由小到大顺序。

判定大小的方式：

 1.被添加元素已实现Comparable接口，通过 compareTo()比较元素大小

 2.指定比较的接口Comparator

非线程安全。

 

有界：（数组长度会自动增长：长度不够时，重建数组，构建器可以指定数组大小）

 

PriorityQueue()
          使用默认的初始容量（11）创建一个 PriorityQueue，并根据其自然顺序对元素进行排序。

PriorityQueue(Collection<? extends E> c)
          创建包含指定 collection 中元素的 PriorityQueue。

PriorityQueue(int initialCapacity)
          使用指定的初始容量创建一个 PriorityQueue，并根据其自然顺序对元素进行排序。

PriorityQueue(int initialCapacity, Comparator<? super E> comparator)
          使用指定的初始容量创建一个 PriorityQueue，并根据指定的比较器对元素进行排序。

PriorityQueue(PriorityQueue<? extends E> c)
          创建包含指定优先级队列元素的 PriorityQueue。

PriorityQueue(SortedSet<? extends E> c)
          创建包含指定有序 set 元素的 PriorityQueue。

 

 

 

 

 

 BlockingQueue 

  在Queue基础上提供了 方法：

1.插入或获取元素时，阻塞直至操作成功。

2.插入或获取元素时，阻塞直至指定时间内操作成功或超时。

JAVA API:

 

boolean	add(E e)           将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则抛出 IllegalStateException。
boolean	contains(Object o)           如果此队列包含指定元素，则返回 true。
int	drainTo(Collection<? super E> c)           移除此队列中所有可用的元素，并将它们添加到给定 collection 中。
int	drainTo(Collection<? super E> c, int maxElements)           最多从此队列中移除给定数量的可用元素，并将这些元素添加到给定 collection 中。
boolean	offer(E e)           将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则返回 false。
boolean	offer(E e, long timeout, TimeUnit unit)           将指定元素插入此队列中，如果队列满，则等待指定时间。在指定时间内可以插入成功则返回true，否则超时返回false
E	poll(long timeout, TimeUnit unit)           获取并移除此队列的头部，如果队列中没有元素，则等待指定时间，在指定时间内可以获得元素就返回该元素，否则超时返回null
void	put(E e)           将指定元素插入此队列中，如果没有可用空间，则阻塞直至插入成功。
int	remainingCapacity()           返回在无阻塞的理想情况下（不存在内存或资源约束）此队列能接受的附加元素数量；如果没有内部限制，则返回 Integer.MAX_VALUE。
boolean	remove(Object o)           从此队列中移除指定元素的单个实例（如果存在）。
E	take()           获取并移除此队列的头部，如果队列中没有元素，则一直阻塞直至能够获取元素。

 

 

ArrayBlockingQueue:

JAVA API描述:

 

 内部采用数组实现的有界阻塞队列。

先进先出队列，新增的元素插入到队列尾部，从头部获取元素。

线程安全实现方式：

  公有方法都使用同一个 ReentrantLock 锁定。

 

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            try {
                while (count == 0)
                    notEmpty.await();//队列空阻塞,直至被插入操作的线程唤醒
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            E x = extract();
            return x;
        } finally {
            lock.unlock();
        }
    }
      private E extract() {
        final E[] items = this.items;
        E x = items[takeIndex];
        items[takeIndex] = null;
        takeIndex = inc(takeIndex);
        --count;
        notFull.signal();//唤醒插入操作线程
        return x;
    }
    
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            try {
                while (count == items.length)
                    notFull.await();//队列满则阻塞,直至被移除操作的线程唤醒
            } catch (InterruptedException ie) {
                notFull.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            insert(e);
        } finally {
            lock.unlock();
        }
    }
   private void insert(E x) {
        items[putIndex] = x;
        putIndex = inc(putIndex);
        ++count;
        notEmpty.signal();//唤醒移除操作的线程
    }
    
    

 内部使用数组实现，初始化时需要指定数组的固定大小，构建器如下：

 

 

ArrayBlockingQueue(int capacity)
          创建一个带有给定的（固定）容量和非公平锁（ReentrantLock ）的 ArrayBlockingQueue。

ArrayBlockingQueue(int capacity, boolean fair)
          创建一个具有给定的（固定）容量和(公平或非公平锁）的 ArrayBlockingQueue。

ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)
          创建一个具有给定的（固定）容量和(公平或非公平锁）的 ArrayBlockingQueue，它最初包含给定 collection 的元素，并以 collection 迭代器的遍历顺序添加元素。

 

 

 

LinkedBlockingQueue:

 

内部采用单向链表实现的有界阻塞队列。（未指定容量时最大容量为Integer.MAX_VALUE,相当于无界。）

先进先出队列，新增的元素插入到队列尾部，从头部获取元素。

 

 

线程安全的，内置两把锁，所有插入操作共用一把锁（护斥），所有的读取操作共用一把锁（护斥），提高了并发性能。 

public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//获取操作锁
        try {
            try {
                while (count.get() == 0)
                    notEmpty.await();//队列为空，阻塞，直至被插入操作的线程唤醒，但被唤醒后再争锁，得到锁时，队列元素可能已被移走即队列扔为空，所以需要加上while(count.get()==0)
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to a non-interrupted thread
                throw ie;
            }

            x = extract();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();//唤醒插入操作线程
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }
    
     public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();//插入操作锁
        try {
            
            try {
                while (count.get() == capacity)
                    notFull.await();//队列满，阻塞，直至被移出操作的线程唤醒，但被唤醒后再争锁，得到锁时，队列可能被其它线程插满了元素即队列扔为满，所以需要加上while(count.get()==capacity)
            } catch (InterruptedException ie) {
                notFull.signal(); // propagate to a non-interrupted thread
                throw ie;
            }
            insert(e);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();//唤醒移出操作的线程 
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

 

 

SynchronousQueue：

 JAVA API:

一种阻塞队列，其中每个插入操作必须等待另一个线程的对应移除操作 ，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。不能在同步队列上进行 peek，因为仅在试图要移除元素时，该元素才存在；除非另一个线程试图移除某个元素，否则也不能（使用任何方法）插入元素；也不能迭代队列，因为其中没有元素可用于迭代。队列的头 是尝试添加到队列中的首个已排队插入线程的元素；如果没有这样的已排队线程，则没有可用于移除的元素并且 poll() 将会返回 null。对于其他 Collection 方法（例如 contains），SynchronousQueue 作为一个空 collection。此队列不允许 null 元素。

 

DelayQueue：延迟队列

内部采用优先级队列(PriorityQueue)实现的无界阻塞队列。 

队列中只能添加Delayed对象，并只有延迟期满时才能从中提取元素。

 Delayed接口：

 

long	getDelay(TimeUnit unit)            返回与此对象相关的剩余延迟时间，以给定的时间单位表示。           如果返回零或负值指示延迟时间已经用尽
int	compareTo(T o)  比较比较对象间的延迟时间，用于排序

 

 

线程安全，所有公用方法都使用同一个 ReentrantLock锁定：

 

    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();//阻塞直至获得到锁
        try {
            E first = q.peek();//取队列头元素
            q.offer(e);//优先级队列使用e的compareTo方法比较，把延迟时间小的排在队列头
            if (first == null || e.compareTo(first) < 0)//之前队列为空，或添加元素的延迟时间小于原队列头原素
                available.signalAll();//唤醒阻塞线程
            return true;
        } finally {
            lock.unlock();//释放锁
        }
    }
    
     public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();//阻塞直至获得到锁
        try {
            E first = q.peek();//取队列头元素
            if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)//队列为空或队列的头原素延迟期未满
                return null;
            else {
                E x = q.poll();//提取队列头元素
                assert x != null;
                if (q.size() != 0)
                    available.signalAll();//唤醒阻塞线程
                return x;
            }
        } finally {
            lock.unlock();//释放锁
        }
    }
    

 

 

PriorityBlockingQueue：

内部采用优先级队列(PriorityQueue)实现的无界有序阻塞队列。

有序：根据添加元素的compareTo方法比较大小或指定的Comparator接口。

无界： PriorityQueue内部采用数组，在构建器里可以指定大小，当队列满时会自动重建数组扩大  容量。

线程安全：所有公用方法都使用同一个 ReentrantLock锁定：

 

 

 

 

双端队列：deque

  deque 接口声明了从队列头部或尾部 插入删 除元素的方法。

 

ArrayDeque:

内部采用数组实现的双端无界队列。

无界：队列满时，会自动重建数组扩大容量。

非线程安全的。

 

将数组看成一个头尾相联的环型。 head 表示当前头下标，tail 表示当前尾下标。

初始时 head=tail =0;

addFirst 方法：head前移添加元素。 

addLast 方法：tail 后移添加元素。

当head==tail说明已闭环（已满）自动重建数组扩大容量。

 源码：

 

    public void addFirst(E e) {
        if (e == null)
            throw new NullPointerException();
        elements[head = (head - 1) & (elements.length - 1)] = e;
         //(head - 1) & (elements.length - 1) 为了实现定位到前一个元素，(elements.length-1) 必须是  二进制数每位全为1的数。 即2^N-1，那么数组的长度就是2^N
         
        //数组看成环形，head为0时，它的前一个就是数组最大下标
        // -1的十六进制数为：0xffffffff,因此-1&任何数X=X;
        //这就实现了数组下标由0移至数组最大下标。
        //
       
        if (head == tail)//闭环了，说明队列已满。
            doubleCapacity();
    }
    
    public void addLast(E e) {
        if (e == null)
            throw new NullPointerException();
        elements[tail] = e;
        if ( (tail = (tail + 1) & (elements.length - 1)) == head)
     
         //(tail + 1) & (elements.length - 1) 为了实现定位到前一个元素，(elements.length-1) 必须是  二进制数每位全为1的数。 即2^N-1，那么数组的长度就是2^N
         //数组看成环形，tail为最大下标时，它的后一个就是数组下标0
         //设数组长度为 2^N ，即2^N-1 & 2^N = 0 
         //这就实现了数组下标由0移至数组最大下标。
        
            doubleCapacity();
    }
    //初始化指定数组大小numElements
    private void allocateElements(int numElements) {
        int initialCapacity = MIN_INITIAL_CAPACITY;
        // Find the best power of two to hold elements.
        // Tests "<=" because arrays aren't kept full.
        if (numElements >= initialCapacity) { //数组实际大小 为大于numElements的最小的2^N
            initialCapacity = numElements;
            initialCapacity |= (initialCapacity >>>  1);
            initialCapacity |= (initialCapacity >>>  2);
            initialCapacity |= (initialCapacity >>>  4);
            initialCapacity |= (initialCapacity >>>  8);
            initialCapacity |= (initialCapacity >>> 16);
            initialCapacity++;

            if (initialCapacity < 0)   // Too many elements, must back off
                initialCapacity >>>= 1;// Good luck allocating 2 ^ 30 elements
        }
        elements = (E[]) new Object[initialCapacity];
    }
    

 

 

LinkedBlockingDeque:

内部采用双向链表实现的有界阻塞双端队列。（未指定容量时最大容量为Integer.MAX_VALUE,相当于无界。）

新增的元素可以插入到队列尾部或头部，也可以从头部或尾部获取元素。

 

 线程安全的，内部使用非公平锁（ReentrantLock）。

所有公用方法都使用同一个 ReentrantLock锁定：

 

 public void put(E e) throws InterruptedException {
	putLast(e);
 }
  public void putLast(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        lock.lock();//阻塞直至获取锁
        try {
            while (!linkLast(e))//无限循环，试着将元素加入队列，只有容量不够时，才失败，成功唤醒移出操作的线程
                notFull.await();//加入队失败时，阻塞，直至被移除操作线程唤醒
        } finally {
            lock.unlock();
        }
    }
    private boolean linkLast(E e) {
        if (count >= capacity)//试着将元素加入队列，只有容量不够时，才失败
            return false;
        ++count;
        Node<E> l = last;
        Node<E> x = new Node<E>(e, l, null);
        last = x;
        if (first == null)
            first = x;
        else
            l.next = x;
        notEmpty.signal();//成功唤醒移出操作的线程
        return true;
    }
    
   public E take() throws InterruptedException {
			return takeFirst();
    }  
    public E takeFirst() throws InterruptedException {
        lock.lock();//阻塞直至获取锁
        try {
            E x;
            while ( (x = unlinkFirst()) == null)//无限循环，试着查找第一个元素，只有容器为空时，返回null，返回非null时唤醒插入操作的线程
                notEmpty.await();//容器为空时，阻塞，直至被插入操作线程唤醒
            return x;
        } finally {
            lock.unlock();
        }
    }  
     private E unlinkFirst() {
        Node<E> f = first;
        if (f == null)//只有容器为空时，返回null
            return null;
        Node<E> n = f.next;
        first = n;
        if (n == null)
            last = null;
        else
            n.prev = null;
        --count;
        notFull.signal();//唤醒插入操作的线程
        return f.item;
    }

 LinkedBlockingDeque 双端队列 提供了在队列两端对元素操作的方法：

 JAVA API:

boolean	add(E e)           在不违反容量限制的情况下，将指定的元素插入此双端队列的末尾。
void	addFirst(E e)           如果立即可行且不违反容量限制，则将指定的元素插入此双端队列的开头；如果当前没有空间可用，则抛出 IllegalStateException。
void	addLast(E e)           如果立即可行且不违反容量限制，则将指定的元素插入此双端队列的末尾；如果当前没有空间可用，则抛出 IllegalStateException。
void	clear()           以原子方式 (atomically) 从此双端队列移除所有元素。
Iterator<E>	descendingIterator()           返回在此双端队列的元素上以逆向连续顺序进行迭代的迭代器。
int	drainTo(Collection<? super E> c)           移除此队列中所有可用的元素，并将它们添加到给定 collection 中。
int	drainTo(Collection<? super E> c, int maxElements)           最多从此队列中移除给定数量的可用元素，并将这些元素添加到给定 collection 中。
E	element()           获取但不移除此双端队列表示的队列的头部。
E	getFirst()           获取，但不移除此双端队列的第一个元素。
E	getLast()           获取，但不移除此双端队列的最后一个元素。
Iterator<E>	iterator()           返回在此双端队列元素上以恰当顺序进行迭代的迭代器。
boolean	offer(E e)           如果立即可行且不违反容量限制，则将指定的元素插入此双端队列表示的队列中（即此双端队列的尾部），并在成功时返回 true；如果当前没有空间可用，则返回 false。
boolean	offer(E e, long timeout, TimeUnit unit)           将指定的元素插入此双端队列表示的队列中（即此双端队列的尾部），必要时将在指定的等待时间内一直等待可用空间。
boolean	offerFirst(E e)           如果立即可行且不违反容量限制，则将指定的元素插入此双端队列的开头，并在成功时返回 true；如果当前没有空间可用，则返回 false。
boolean	offerFirst(E e, long timeout, TimeUnit unit)           将指定的元素插入此双端队列的开头，必要时将在指定的等待时间内等待可用空间。
boolean	offerLast(E e)           如果立即可行且不违反容量限制，则将指定的元素插入此双端队列的末尾，并在成功时返回 true；如果当前没有空间可用，则返回 false。
boolean	offerLast(E e, long timeout, TimeUnit unit)           将指定的元素插入此双端队列的末尾，必要时将在指定的等待时间内等待可用空间。
E	peek()           获取但不移除此双端队列表示的队列的头部（即此双端队列的第一个元素）；如果此双端队列为空，则返回 null。
E	peekFirst()           获取，但不移除此双端队列的第一个元素；如果此双端队列为空，则返回 null。
E	peekLast()           获取，但不移除此双端队列的最后一个元素；如果此双端队列为空，则返回 null。
E	poll()           获取并移除此双端队列表示的队列的头部（即此双端队列的第一个元素）；如果此双端队列为空，则返回 null。
E	poll(long timeout, TimeUnit unit)           获取并移除此双端队列表示的队列的头部（即此双端队列的第一个元素），如有必要将在指定的等待时间内等待可用元素。
E	pollFirst()           获取并移除此双端队列的第一个元素；如果此双端队列为空，则返回 null。
E	pollFirst(long timeout, TimeUnit unit)           获取并移除此双端队列的第一个元素，必要时将在指定的等待时间等待可用元素。
E	pollLast()           获取并移除此双端队列的最后一个元素；如果此双端队列为空，则返回 null。
E	pollLast(long timeout, TimeUnit unit)           获取并移除此双端队列的最后一个元素，必要时将在指定的等待时间内等待可用元素。
E	pop()           从此双端队列所表示的堆栈中弹出一个元素。
void	push(E e)           将元素推入此双端队列表示的栈。
void	put(E e)           将指定的元素插入此双端队列表示的队列中（即此双端队列的尾部），必要时将一直等待可用空间。
void	putFirst(E e)           将指定的元素插入此双端队列的开头，必要时将一直等待可用空间。
void	putLast(E e)           将指定的元素插入此双端队列的末尾，必要时将一直等待可用空间。
int	remainingCapacity()           返回理想情况下（没有内存和资源约束）此双端队列可不受阻塞地接受的额外元素数。
E	remove()           获取并移除此双端队列表示的队列的头部。
boolean	remove(Object o)           从此双端队列移除第一次出现的指定元素。
E	removeFirst()           获取并移除此双端队列第一个元素。
boolean	removeFirstOccurrence(Object o)           从此双端队列移除第一次出现的指定元素。
E	removeLast()           获取并移除此双端队列的最后一个元素。
boolean	removeLastOccurrence(Object o)           从此双端队列移除最后一次出现的指定元素。
E	takeFirst()           获取并移除此双端队列的第一个元素，必要时将一直等待可用元素。
E	takeLast()           获取并移除此双端队列的最后一个元素，必要时将一直等待可用元素。