并发队列之ConcurrentLinkedQueue

Posted wyq1995

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了并发队列之ConcurrentLinkedQueue相关的知识,希望对你有一定的参考价值。

  本来想着直接说线程池的,不过在说线程池之前,我们必须要知道并发安全队列;因为一般情况下线程池中的线程数量是一定的,肯定不会超过某个阈值,那么当任务太多了的时候,我们必须把多余的任务保存到并发安全队列中,当线程池中的线程空闲下来了,就会到并发安全队列中拿任务;

  那么什么是并发安全队列呢?其实可以简单看作是一个链表,然后我们先办法去存取节点;总的来说,并发安全队列分为两种,一种是阻塞的,一种是非阻塞的,前者是用锁来实现的,后者用CAS实现的;

 

一.简单介绍ConcurrentLinkedQueue

  这个队列用法没什么好说的,就类似LinkedList的用法,怎么对一个链表继续增删改查,不多说,我们就说一下其中几个关键的方法;

  首先,这个队列是一个线程安全的无界非阻塞队列,其实就是一个单向链表,无界的意思就是没有限制最大长度,非阻塞表示用CAS实现入队和出队操作,我们打开这个类就可以知道,有一个内部类Node,其中重要的属性如下所示:

技术图片

//用于存放节点的值
volatile E item;
//指向下一个节点
volatile Node<E> next;
//这里也是用的是UNSAFE类,前面说过了,这个类直接提供CAS操作
private static final sun.misc.Unsafe UNSAFE;
//item字段的偏移量
private static final long itemOffset;
//next的偏移量
private static final long nextOffset;

 

 

  然后ConcurrentLinkedQueue中几个重要的属性,好像也没什么重要的,就保存了头节点和尾节点,注意,默认情况下头节点和尾节点都是哨兵节点,也就是一个存null的Node节点

//存放链表的头节点
private transient volatile Node<E> head;
//存放链表的尾节点
private transient volatile Node<E> tail;
//UNSAFE对象
private static final sun.misc.Unsafe UNSAFE;
//head字段的偏移量
private static final long headOffset;
//tail字段偏移量
private static final long tailOffset;

 

技术图片

 

 

 

  下面我们直接看一些重要方法吧!慢慢分析其中的算法才是关键的

 

二.offer方法

  这个方法的作用就是在队列末端添加一个节点,如果传递的参数是null,就抛出空指针异常,否则由于该队列是无界队列,该方法会一直返回true,而且该方法使用CAS算法实现的,所以不会阻塞线程;

//队列末端添加一个节点
public boolean offer(E e) {
    //如果e为空,那么抛出空指针异常
    checkNotNull(e);
    //将传进来的元素封装成一个节点,Node的构造器中调用UNSAFE.putObject(this, itemOffset, item)把e赋值给节点中的item
    final Node<E> newNode = new Node<E>(e);

    //[1]
    //这里的for循环从最后的节点开始
    for (Node<E> t = tail, p = t;;) {
      Node<E> q = p.next;
      //[2]如果q为null,说明p就是最后的节点了
        if (q == null) {
            //[3]CAS更新:如果p节点的下一个节点是null,就把写个节点更新为newNode
            if (p.casNext(null, newNode)) {
                //[4]CAS成功,但是这时p==t,所以不会进入到这里的if里面,直接返回true
                //那么什么时候会走到这里面来呢?其实是要有另外一个线程也在调用offer方法的时候,会进入到这里面来
                if (p != t) 
                    casTail(t, newNode);  
                return true;
            }
        }
        else if (p == q) //[5]
            
            p = (t != (t = tail)) ? t : head;
        else //[6]
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

 

  

  上面执行到[3]的时候,由于头节点和尾节点默认都是指向哨兵节点的,由于这个时候p的下一个节点为null,所以当前线程A执行CAS会成功,下图所示;

技术图片

 

 

  如果此时还有一个线程B也来尝试[3]中CAS,由于此时p节点的下一个节点不是null了,于是线程B会跳到[1]出进行第二次循环,然后会到[6]中,由于p和t此时是相等的,所以这里是false,即p=q,下图所示:

技术图片

 

 

  然后线程B又会跳到[1]处进行第三次循环,由于执行了Node<E> q = p.next,所以此时q指向最后的null,就到了[3]处进行CAS,这次是可以成功的,成功之后如下图所示:

 技术图片

 

   

  这个时候因为p!=t,所以可以进入到[4],这里又会进行一个CAS:如果tail和t指向的节点一样,那么就将tail指向新添加的节点,如图所示,这个时候线程B也就执行完了;

技术图片

 

   

  其实还有[5]没有走到,这个是在poll操作之后才执行的,我们先跳过,等说完poll方法之后再回头看看;另外说一下,add方法其实就是调用的是offer方法,就不多说了;

技术图片

 

 

三.poll方法

  这个方法是获取头部的这个节点,如果队列为空则返回null;

public E poll() {
    //这里其实就是一个goto的标记,用于跳出for循环
    restartFromHead:
    //[1]
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;
            //[2]如果当前节点中存的值不为空,则CAS设置为null
            if (item != null && p.casItem(item, null)) {
                //[3]CAS成功就更新头节点的位置
                if (p != h) 
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //[4]当前队列为空,就返回null
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //[5]当前节点和下一个节点一样,说明节点自引用,则重新找头节点
            else if (p == q)
                continue restartFromHead;
            //[6]
            else
                p = q;
        }
    }
}

final void updateHead(Node<E> h, Node<E> p) {
    if (h != p && casHead(h, p))
        h.lazySetNext(h);
}

  

  分为几种情况,第一种情况是线程A调用poll方法的时候,发现队列是空的,即头节点和尾节点都指向哨兵节点,就会直接到[4],返回null

  第二种情况,线程A执行到了[4],此时有一个线程却调用offer方法添加了一个节点,下图所示,那么此时线程A就不会走[4]了,[5]也不满足,于是会到[6]这里来,然后线程A又会跳到[1]处进行循环,此时p指向的节点中item不为null,所以会到[2]中;

技术图片

 

   

  到了[2]中将p指向的节点中item用CAS设置为null,然后就到了[3],下面第一个图,由于p!=h,q=null,所以最后调用的是updateHead(h,p),这方法:如果头节点和h指向的是一样的,就将头节点指向p,我们还能看到updateHead方法中h.lazySetNext(h)表示h的下一个节点指向自己,下面图二

技术图片

 

   到了这里还没完,还记不记得offer方法中有一个地方的代码没有执行的啊!就是这种情况,尾节点自己引用自己,我们再调用offer会怎么样呢?

  回到offer方法,先会到[1],然后q指向自己这个哨兵节点(注意,此时虽然p指向的节点中存的是null,但是p!=null},于是再到[5],此时的图如下左图所示;此时由于t==tail,所以p=head;

技术图片

 

   再在offer方法循环一次,此时q指向null,下面左图所示,然后就可以进入[2]中进行CAS,CAS成功,因为此时p!=t,所以还要进行CAS将tail指向新节点,下面右图所示,可以让GC回收那个垃圾!

妈耶,这里比较绕!哈哈哈哈哈哈哈哈哈哈哈

 技术图片

 

 

四.peek方法

  这个方法的作用就是获取队列头部的元素,只获取不移除,注意这个方法和上面的poll方法的区别啊!

public E peek() {
    //[1]goto标志
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            //[2]
            E item = p.item;
            //[3]
            if (item != null || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            }
            //[4]
            else if (p == q)
                continue restartFromHead;
            else//[5]
                p = q;
        }
    }
}

  final void updateHead(Node<E> h, Node<E> p) {
    if (h != p && casHead(h, p))
        h.lazySetNext(h);
  }

 

 

  

  如果队列中为空的时候,走到[3]的时候,就会如下图所示,由于h==p,所以updateHead方法啥也不做,然后返回就返回item为null

技术图片

 

 

  如果队列不为空,那么如下左图所示,此时进入循环内不满足条件,会到[5]这里,将p指向q,然后再进行一次循环到[3],将q指向p的后一个节点,下面右图所示;

技术图片

 

 

  然后调用updateHead方法,用CAS将头节点指向p这里,然后将h自己指向自己,下图所示,最后返回item

技术图片

 

 

五.总结

  其实还有几个方法没说,但是感觉比较容易就不浪费篇幅了,有兴趣的可以看看:size方法用于计算队列中节点的数量,可是由于没有加锁,在并发的条件下不准确;remove方法删除某个节点,其实就是遍历然后用equals方法比较item是不是一样,只不过如果存在多个符合条件的节点只删除第一个,然后返回true,否则返回false;contains方法判断队列中是否包含指定item的节点,也就是遍历,很容易;

  最麻烦的就是offer方法和poll方法,offer方法是在队列的最后面添加节点,而poll是获取头节点,并且删除第一个真正的队列节点(注意,节点分为两种,一种是哨兵节点,一种是真正的存了数据的节点啊),还简单的说了一下poll方法和peek方法的区别,后者只是获取,而不删除啊!用下面这个图帮助记忆一下;

技术图片

 

以上是关于并发队列之ConcurrentLinkedQueue的主要内容,如果未能解决你的问题,请参考以下文章

测开之并发编程篇・《并发并行线程队列进程和协程》

JUC系列并发容器之阻塞队列Overview

并发队列之ArrayBlockingQueue

Python 3 并发编程多进程之队列(推荐使用)

java并发之阻塞队列

高并发框架之消息队列