【java】Java多线程总结之线程安全队列Queue【转载】

原文地址：https://www.cnblogs.com/java-jun-world2099/articles/10165949.html

 

========================================================================

在Java多线程应用中，队列的使用率很高，多数生产消费模型的首选数据结构就是队列。Java提供的线程安全的Queue能够分为阻塞队列和非阻塞队列，其中阻塞队列的典型例子是BlockingQueue，非阻塞队列的典型例子是

ConcurrentLinkedQueue，在实际应用中要根据实际须要选用阻塞队列或者非阻塞队列。

注：什么叫线程安全？这个首先要明确。线程安全的类 ，指的是类内共享的全局变量的访问必须保证是不受多线程形式影响的。若是因为多线程的访问（好比修改、遍历、查看）而使这些变量结构被破坏或者针对这些变量操做的原子性被破坏，则这个类就不是线程安全的。

在并发的队列上jdk提供了两套实现，一个是以ConcurrentLinkedQueue为表明的高性能队列，一个是以BlockingQueue接口为表明的阻塞队列，不管在那种都继承自Queue。
今天就聊聊这两种Queue：

• BlockingQueue  阻塞算法
• ConcurrentLinkedQueue，非阻塞算法

1、首先来看看BlockingQueue： 

Queue是什么就不须要多说了吧，一句话：队列是先进先出。相对的，栈是后进先出。若是不熟悉的话先找本基础的数据结构的书看看吧。 

BlockingQueue，顾名思义，“阻塞队列”：能够提供阻塞功能的队列。 
首先，看看BlockingQueue提供的经常使用方法： 

从上表能够很明显看出每一个方法的做用，这个不用多说。我想说的是：

• add(e) remove() element() 方法不会阻塞线程。当不知足约束条件时，会抛出IllegalStateException 异常。例如：当队列被元素填满后，再调用add(e)，则会抛出异常。
• offer(e) poll() peek() 方法即不会阻塞线程，也不会抛出异常。例如：当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
• 要想要实现阻塞功能，须要调用put(e) take() 方法。当不知足约束条件时，会阻塞线程。

• BlockingQueue  阻塞算法

BlockingQueue做为线程容器，能够为线程同步提供有力的保障。

BlockingQueue定义的经常使用方法：

　　   抛出异常    特殊值      阻塞       超时

插入   add(e)        offer(e)      put(e)     offer(e, time, unit)
移除   remove()    poll()         take()      poll(time, unit)
检查   element()   peek()       不可用    不可用

一、ArrayBlockingQueue

　　基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个经常使用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。
　　ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着二者没法真正并行运行，这点尤为不一样于LinkedBlockingQueue；按照实现原理来分析，ArrayBlockingQueue彻底能够采用分离锁，从而实现生产者和消费者操做的彻底并行运行。Doug Lea之因此没这样去作，也许是由于ArrayBlockingQueue的数据写入和获取操做已经足够轻巧，以致于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上彻底占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不一样之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，然后者则会生成一个额外的Node对象。这在长时间内须要高效并发地处理大批量数据的系统中，其对于GC的影响仍是存在必定的区别。而在建立ArrayBlockingQueue时，咱们还能够控制对象的内部锁是否采用公平锁，默认采用非公平锁。

二、LinkedBlockingQueue

　　基于链表的阻塞队列，同ArrayListBlockingQueue相似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者当即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue能够经过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于一样的原理。而LinkedBlockingQueue之因此可以高效的处理并发数据，还由于其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的状况下生产者和消费者能够并行地操做队列中的数据，以此来提升整个队列的并发性能。
做为开发者，咱们须要注意的是，若是构造一个LinkedBlockingQueue对象，而没有指定其容量大小，LinkedBlockingQueue会默认一个相似无限大小的容量（Integer.MAX_VALUE），这样的话，若是生产者的速度一旦大于消费者的速度，也许尚未等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

阻塞队列：线程安全

　　按 FIFO（先进先出）排序元素。队列的头部 是在队列中时间最长的元素。队列的尾部 是在队列中时间最短的元素。新元素插入到队列的尾部，而且队列检索操做会得到位于队列头部的元素。连接队列的吞吐量一般要高于基于数组的队列，可是在大多数并发应用程序中，其可预知的性能要低。

注意：

一、必需要使用take()方法在获取的时候达成阻塞结果
二、使用poll()方法将产生非阻塞效果

三、LinkedBlockingQueue实例

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

 

public class BlockingDeque {
    //阻塞队列，FIFO
    private static LinkedBlockingQueue<Integer> concurrentLinkedQueue = new LinkedBlockingQueue<Integer>(); 

          
 public static void main(String[] args) {  
     ExecutorService executorService = Executors.newFixedThreadPool(2);  

     executorService.submit(new Producer("producer1"));  
     executorService.submit(new Producer("producer2"));  
     executorService.submit(new Producer("producer3"));  
     executorService.submit(new Consumer("consumer1"));  
     executorService.submit(new Consumer("consumer2"));  
     executorService.submit(new Consumer("consumer3"));  

 }  

 static class Producer implements Runnable {  
     private String name;  

     public Producer(String name) {  
         this.name = name;  
     }  

     public void run() {  
         for (int i = 1; i < 10; ++i) {  
             System.out.println(name+ "  生产： " + i);  
             //concurrentLinkedQueue.add(i);  
             try {
                concurrentLinkedQueue.put(i);
                Thread.sleep(200); //模拟慢速的生产，产生阻塞的效果
            } catch (InterruptedException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }
             
         }  
     }  
 }  

 static class Consumer implements Runnable {  
     private String name;  

     public Consumer(String name) {  
         this.name = name;  
     }  
     public void run() {  
         for (int i = 1; i < 10; ++i) {  
             try {          
                    //必需要使用take()方法在获取的时候阻塞
                      System.out.println(name+"消费： " +  concurrentLinkedQueue.take());  
                      //使用poll()方法 将产生非阻塞效果
                      //System.out.println(name+"消费： " +  concurrentLinkedQueue.poll());  
                     
                     //还有一个超时的用法，队列空时，指定阻塞时间后返回，不会一直阻塞
                     //但有一个疑问，既然能够不阻塞，为啥还叫阻塞队列？
                    //System.out.println(name+" Consumer " +  concurrentLinkedQueue.poll(300, TimeUnit.MILLISECONDS));                    
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }  

         }  
     }  
 }  
}

四、PriorityBlockingQueue

基于优先级的阻塞队列（优先级的判断经过构造函数传入的Compator对象来决定，也就是说传入队列的对象必须实现Comparable接口），在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁，他也是一个无界的队列。

五、PriorityBlockingQueue 实例

Task.java

  View Code

UsePriorityBlockingQueue.java

  View Code

打印结果：

  View Code

六、DelayQueue

带有延迟时间的Queue，其中的元素只有当其指定的延迟时间到了，才可以从队列中获取到该元素。DelayQueue中的元素必须实现Delayed接口，DelayQueue是一个没有大小限制的队列，应用场景不少，好比对缓存超时的数据进行移除、任务超时处理、空闲链接的关闭等等。

七、DelayQueue实例

Wangmin.java

  View Code

WangBa.java

  View Code

打印结果：

  View Code

八、LinkedBlockingDeque

　　LinkedBlockingDeque是一个线程安全的双端队列实现，由链表结构组成的双向阻塞队列，便可以从队列的两端插入和移除元素。双向队列由于多了一个操做队列的入口，在多线程同时入队时，也就减小了一半的竞争。能够说他是最为复杂的一种队列，在内部实现维护了前端和后端节点，可是其没有实现读写分离，所以同一时间只能有一个线程对其讲行操做。在高并发中性能要远低于其它BlockingQueue。更要低于ConcurrentLinkedQueue，jdk早期有一个非线程安全的Deque就是ArryDeque了， java6里添加了LinkBlockingDeque来弥补多线程场景下线程安全的问题。

相比于其余阻塞队列，LinkedBlockingDeque多了addFirst、addLast、peekFirst、peekLast等方法，以first结尾的方法，表示插入、获取获移除双端队列的第一个元素。以last结尾的方法，表示插入、获取获移除双端队列的最后一个元素。

此外，LinkedBlockingDeque仍是可选容量的,防止过分膨胀，默认等于Integer.MAX_VALUE。

主要方法：

　　akeFirst()和takeLast()：分别返回类表中第一个和最后一个元素，返回的元素会从类表中移除。若是列表为空，调用的方法的线程将会被阻塞直达列表中有可用元素。

　　getFirst()和getLast():分别返回类表中第一个和最后一个元素，返回的元素不会从列表中移除。若是列表为空，则抛出NoSuckElementException异常。

　　peek()、peekFirst()和peekLast():分别返回列表中第一个元素和最后一个元素，返回元素不会被移除。若是列表为空返回null。

　　poll()、pollFirst()和pollLast():分别返回类表中第一个和最后一个元素，返回的元素会从列表中移除。若是列表为空，返回Null。

public class UseDeque {
    public static void main(String[] args) {
        LinkedBlockingDeque<String> dq = new LinkedBlockingDeque<String>(10);
        dq.addFirst("a");
        dq.addFirst("b");
        dq.addFirst("c");
        dq.addFirst("d");
        dq.addFirst("e");
        dq.addLast("f");
        dq.addLast("g");
        dq.addLast("h");
        dq.addLast("i");
        dq.addLast("j");
        //dq.offerFirst("k");
        System.out.println("查看头元素：" + dq.peekFirst());
        System.out.println("获取尾元素：" + dq.pollLast());
        Object [] objs = dq.toArray();
        for (int i = 0; i < objs.length; i++) {
            System.out.print(objs[i] + " -- ");
        }
    }
}

打印结果：

  View Code

九、LinkedBlockingDeque方法列表

  View Code

十、LinkedBlockingQeque和LinkedBlockingDeque源码解读

1）LinkedBlockingQeque

先看它的结构基本字段：

  View Code

和LinkedBlockingDeque的区别之一就是，LinkedBlockingQueue采用了两把锁来对队列进行操做，也就是队尾添加的时候， 

队头仍然能够删除等操做。接下来看典型的操做。

put操做

  View Code

主要的思想仍是比较容易理解的，如今看看enqueue 方法：

private void enqueue(Node<E> node) {        //入对操做。
        last = last.next = node;      //队尾进
}

再看看signalNotEmpty方法：

private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();        //加锁
        try {
            notEmpty.signal();    //用于signal，notEmpty
        } finally {
            takeLock.unlock();
        }
}

take操做

take操做，就是从队列里面弹出一个元素，下面看它的详细代码：

public E take() throws InterruptedException {
        E x;
        int c = -1;            //设定一个记录变量
        final AtomicInteger count = this.count;     //得到count
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();        //加锁
        try {
            while (count.get() == 0) {       //若是没有元素，那么就阻塞性等待
                notEmpty.await();
            }
            x = dequeue();            //必定能够拿到。
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();        //报告还有元素，唤醒队列
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();           //解锁
        return x;
}

接下来看dequeue方法：

private E dequeue() {
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h;        // help GC 指向本身，帮助gc回收
        head = first;
        E x = first.item;       //从队头出。
        first.item = null;      //将head.item设为null。
        return x;
}

对于LinkedBlockingQueue来讲，有两个ReentrantLock分别控制队头和队尾，这样就可使得添加操做分开来作，通常的操做是获取一把锁就能够，但有些操做例如remove操做，则须要同时获取两把锁：

public boolean remove(Object o) {
        if (o == null) return false;
        fullyLock();     //获取锁
        try {
            for (Node<E> trail = head, p = trail.next;
                 p != null;
                 trail = p, p = p.next) {     //依次循环遍历
                if (o.equals(p.item)) {       //找到了
                    unlink(p, trail);       //解除连接
                    return true;
                }
            }
            return false;        //没找到，或者解除失败
        } finally {
            fullyUnlock();
        }
}

固然，除了上述的remove方法外，在Iterator的next方法，remove方法以及LBQSpliterator分割迭代器中也是须要加全锁进行操做的。

2）LinkedBlockingDeque

LinkedBlockingDeque类有三个构造方法：

public LinkedBlockingDeque()
public LinkedBlockingDeque(int capacity)
public LinkedBlockingDeque(Collection<? extends E> c)

LinkedBlockingDeque类中的数据都被封装成了Node对象：

static final class Node<E> {
    E item;
    Node<E> prev;
    Node<E> next;
 
    Node(E x) {
        item = x;
    }
}

LinkedBlockingDeque类中的重要字段以下：

// 队列双向链表首节点
transient Node<E> first;
// 队列双向链表尾节点
transient Node<E> last;
// 双向链表元素个数
private transient int count;
// 双向链表最大容量
private final int capacity;
// 全局独占锁
final ReentrantLock lock = new ReentrantLock();
// 非空Condition对象
private final Condition notEmpty = lock.newCondition();
// 非满Condition对象
private final Condition notFull = lock.newCondition();

LinkedBlockingDeque类的底层实现和LinkedBlockingQueue类很类似，都有一个全局独占锁，和两个Condition对象，用来阻塞和唤醒线程。

LinkedBlockingDeque类对元素的操做方法比较多，咱们下面以putFirst、putLast、pollFirst、pollLast方法来对元素的入队、出队操做进行分析。

入队

putFirst(E e)方法是将指定的元素插入双端队列的开头，源码以下：

public void putFirst(E e) throws InterruptedException {
    // 若插入元素为null，则直接抛出NullPointerException异常
    if (e == null) throw new NullPointerException();
    // 将插入节点包装为Node节点
    Node<E> node = new Node<E>(e);
    // 获取全局独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        while (!linkFirst(node))
            notFull.await();
    } finally {
        // 释放全局独占锁
        lock.unlock();
    }
}

入队操做是经过linkFirst(E e)方法来完成的，以下所示：

private boolean linkFirst(Node<E> node) {
    // assert lock.isHeldByCurrentThread();
    // 元素个数超出容量。直接返回false
    if (count >= capacity)
        return false;
    // 获取双向链表的首节点
    Node<E> f = first;
    // 将node设置为首节点
    node.next = f;
    first = node;
    // 若last为null，设置尾节点为node节点
    if (last == null)
        last = node;
    else
        // 更新原首节点的前驱节点
        f.prev = node;
    ++count;
    // 唤醒阻塞在notEmpty上的线程
    notEmpty.signal();
    return true;
}

若入队成功，则linkFirst(E e)方法返回true，不然，返回false。若该方法返回false，则当前线程会阻塞在notFull条件上。

putLast(E e)方法是将指定的元素插入到双端队列的末尾，源码以下：

public void putLast(E e) throws InterruptedException {
    // 若插入元素为null，则直接抛出NullPointerException异常
    if (e == null) throw new NullPointerException();
    // 将插入节点包装为Node节点
    Node<E> node = new Node<E>(e);
    // 获取全局独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        while (!linkLast(node))
            notFull.await();
    } finally {
        // 释放全局独占锁
        lock.unlock();
    }
}

该方法和putFirst(E e)方法几乎同样，不一样点在于，putLast(E e)方法经过调用linkLast(E e)方法来插入节点：

private boolean linkLast(Node<E> node) {
    // assert lock.isHeldByCurrentThread();
    // 元素个数超出容量。直接返回false
    if (count >= capacity)
        return false;
    // 获取双向链表的尾节点
    Node<E> l = last;
    // 将node设置为尾节点
    node.prev = l;
    last = node;
    // 若first为null，设置首节点为node节点
    if (first == null)
        first = node;
    else
        // 更新原尾节点的后继节点
        l.next = node;
    ++count;
    // 唤醒阻塞在notEmpty上的线程
    notEmpty.signal();
    return true;
}

若入队成功，则linkLast(E e)方法返回true，不然，返回false。若该方法返回false，则当前线程会阻塞在notFull条件上。

出队

pollFirst()方法是获取并移除此双端队列的首节点，若不存在，则返回null，源码以下：

public E pollFirst() {
    // 获取全局独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return unlinkFirst();
    } finally {
        // 释放全局独占锁
        lock.unlock();
    }
}

移除首节点的操做是经过unlinkFirst()方法来完成的：

private E unlinkFirst() {
    // assert lock.isHeldByCurrentThread();
    // 获取首节点
    Node<E> f = first;
    // 首节点为null，则返回null
    if (f == null)
        return null;
    // 获取首节点的后继节点
    Node<E> n = f.next;
    // 移除first，将首节点更新为n
    E item = f.item;
    f.item = null;
    f.next = f; // help GC
    first = n;
    // 移除首节点后，为空队列
    if (n == null)
        last = null;
    else
        // 将新的首节点的前驱节点设置为null
        n.prev = null;
    --count;
    // 唤醒阻塞在notFull上的线程
    notFull.signal();
    return item;
}

 

pollLast()方法是获取并移除此双端队列的尾节点，若不存在，则返回null，源码以下：

public E pollLast() {
    // 获取全局独占锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return unlinkLast();
    } finally {
        // 释放全局独占锁
        lock.unlock();
    }
}

移除尾节点的操做是经过unlinkLast()方法来完成的：

private E unlinkLast() {
    // assert lock.isHeldByCurrentThread();
    // 获取尾节点
    Node<E> l = last;
    // 尾节点为null，则返回null
    if (l == null)
        return null;
    // 获取尾节点的前驱节点
    Node<E> p = l.prev;
    // 移除尾节点，将尾节点更新为p
    E item = l.item;
    l.item = null;
    l.prev = l; // help GC
    last = p;
    // 移除尾节点后，为空队列
    if (p == null)
        first = null;
    else
        // 将新的尾节点的后继节点设置为null
        p.next = null;
    --count;
    // 唤醒阻塞在notFull上的线程
    notFull.signal();
    return item;
}

其实LinkedBlockingDeque类的入队、出队操做都是经过linkFirst、linkLast、unlinkFirst、unlinkLast这几个方法来实现的，源码读起来也比较简单。

2、ConcurrentLinkedQueue 非阻塞算法

一、非阻塞队列

基于连接节点的、无界的、线程安全。此队列按照 FIFO（先进先出）原则对元素进行排序。队列的头部 是队列中时间最长的元素。队列的尾部 是队列中时间最短的元素。新的元素插入到队列的尾部，队列检索操做从队列头部得到元素。当许多线程共享访问一个公共 collection 时，ConcurrentLinkedQueue 是一个恰当的选择。此队列不容许 null 元素。

ConcurrentLinkedQueue是一个适用于高并发场景下的队列，经过无锁的方式，实现了高并发状态下的高性能，一般ConcurrentLinkedQueue性能好于BlockingQueue。

ConcurrentLinkedQueue重要方法：

add()和offer()都是加入元素的方法（在ConcurrentLinkedQueue中，这两个方法投有任何区别）

poll()和peek()都是取头元素节点，区别在于前者会删除元素，后者不会，至关于查看。

public class UseQueue_ConcurrentLinkedQueue {


    public static void main(String[] args) throws Exception {

        //高性能无阻塞无界队列：ConcurrentLinkedQueue

        ConcurrentLinkedQueue<String> q = new ConcurrentLinkedQueue<String>();
        q.offer("a");
        q.offer("b");
        q.offer("c");
        q.offer("d");
        q.add("e");

        System.out.println("从头部取出元素，并从队列里删除 >> "+q.poll());    //a 从头部取出元素，并从队列里删除
        System.out.println("删除后的长度 >> "+q.size());    //4
        System.out.println("取出头部元素 >> "+q.peek());    //b
        System.out.println("长度 >> "+q.size());    //4
        }
}

打印结果：

  View Code

二、实例

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;


public class NoBlockQueue {  
       private static ConcurrentLinkedQueue<Integer> concurrentLinkedQueue = new ConcurrentLinkedQueue<Integer>();   
          
    public static void main(String[] args) {  
        ExecutorService executorService = Executors.newFixedThreadPool(2);  

        executorService.submit(new Producer("producer1"));  
        executorService.submit(new Producer("producer2"));  
        executorService.submit(new Producer("producer3"));  
        executorService.submit(new Consumer("consumer1"));  
        executorService.submit(new Consumer("consumer2"));  
        executorService.submit(new Consumer("consumer3"));  

    }  
  
    static class Producer implements Runnable {  
        private String name;  
  
        public Producer(String name) {  
            this.name = name;  
        }  
  
        public void run() {  
            for (int i = 1; i < 10; ++i) {  
                System.out.println(name+ " start producer " + i);  
                concurrentLinkedQueue.add(i);  
                try {
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                //System.out.println(name+"end producer " + i);  
            }  
        }  
    }  
  
    static class Consumer implements Runnable {  
        private String name;  
  
        public Consumer(String name) {  
            this.name = name;  
        }  
        public void run() {  
            for (int i = 1; i < 10; ++i) {  
                try {
 
                    System.out.println(name+" Consumer " +  concurrentLinkedQueue.poll());

                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }  
//                System.out.println();  
//                System.out.println(name+" end Consumer " + i);  
            }  
        }  
    }  
} 

　　在并发编程中，通常推荐使用阻塞队列，这样实现能够尽可能地避免程序出现意外的错误。阻塞队列使用最经典的场景就是socket客户端数据的读取和解析，读取数据的线程不断将数据放入队列，而后解析线程不断从队列取数据解析。还有其余相似的场景，只要符合生产者-消费者模型的均可以使用阻塞队列。

使用非阻塞队列，虽然能即时返回结果（消费结果），但必须自行编码解决返回为空的状况处理（以及消费重试等问题）。

另外它们都是线程安全的，不用考虑线程同步问题。

3、多线程模拟队列

  View Code

 

 

 

=====================================

留做参考文章