Java并发编程 ConcurrentLinkedQueue

时间:2022-05-28 派大大大星人气:0

ConcurrentLinkedQueue

JDK中提供了一系列场景的并发安全队列。总的来说，按照实现方式的不同可分为阻塞队列和非阻塞队列，前者使用锁实现，而后则使用CAS非阻塞算法实现。

ConcurrentLinkedQueue 内部的队列使用单向链表方式实现，其中有两个volatile 类型的 Node 节点分别用来存放队列的首、尾节点。从下面的无参构造函数可知，默认头、尾节点都是指向 item 为null 的哨兵节点。新元素会被插入队列末尾，出队时从队列头部获取一个元素。

public ConcurrentLinkedQueue() {
    head = tail = new Node<E>(null);
}

在 Node 节点内部则维护一个使用volatile 修饰的变量 item，用来存放节点的值;next用来存放链表的下一个节点，从而链接为一个单向无界链表。其内部则使用 UNSafe 工具类提供的CAS 算法来保证出入队时操作链表的原子性。

下面通过介绍ConcurrentLinkedQueue的几个方法来介绍其实现原理。

offer操作： offer操作是在队列末尾添加一个元素，如果传递的参数是null则抛出NPE异常，否则由于ConcurrentLinkedQueue是无界队列，该方法一直会返回true。另外，由于使用CAS无阻塞算法，因此该方法不会阻塞挂起调用线程。下面具体看下实现原理。

public boolean offer(E e) {
//（1）e为null这抛出空指针异常
    checkNotNull(e);
    //(2)构造Node节点，在构造函数内部调用unsafe.putObject
    final Node<E> newNode = new Node<E>(e);
    //(3) 从尾节点插入
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
 //(4) 如果q==null说明p是尾节点，则执行插入
        if (q == null) {
            // p is last node
            //（5）使用CAS设置p节点的next节点
            if (p.casNext(null, newNode)) {
                // Successful CAS is the linearization point
                // for e to become an element of this queue,
                // and for newNode to become "live".
                   //（6）CAS成功，则说明新增节点已经放入链表，然后设置当前尾巴节点

                if (p != t) // hop two nodes at a time
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
        else if (p == q)
            // We have fallen off list.  If tail is unchanged, it
            // will also be off-list, in which case we need to
            // jump to head, from which all live nodes are always
            // reachable.  Else the new tail is a better bet.
            p = (t != (t = tail)) ? t : head;
        else
            // Check for tail updates after two hops.
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

首先看当一个线程调用offer(item)时的情况。首先代码(1)对传参进行空检查，由于使用如果为null 则抛出NPE 异常，否则执行代码(2)并使用item作为构造函数参数创建一个新的节点，然后代码(3)从队列尾部节点开始循环，打算从队列尾部添加元素。这时候节点p、t、head、tail同时指向了item为null的哨兵节点，由于哨兵节点的next 节点为null，所以这里q也指向null。代码(4)发现q->null则执行代码(5),通过CAS 原子操作判断p节点的next节点是否为null，如果为null 则使用节点newNode替换p的next节点，然后执行代码(6)，这里由于p=t所以没有设置尾部节点，然后退出 offer方法。
上面是一个线程调用offer方法的情况，如果多个线程同时调用，就会存在多个线程同时执行到代码(5)的情况。假设线程A调用offer(item1)，线程B调用 ofer(item2)，同时执行到代码(5)p.casNext(null, newNode)。由于CAS的比较设置操作是原子性的，所以这里假设线程A先执行了比较设置操作，发现当前p的 next 节点确实是null，则会原子性地更新next节点为iteml，这时候线程B也会判断p的next节点是否为null，结果发现不是null(因为线程A已经设置了p的next节点为iteml)，则会跳到代码(3)，然后执行到代码(4)。

可见，offer 操作中的关键步骤是代码(5)，通过原子CAS 操作来控制某时只有一个线程可以追加元素到队列末尾。进行CAS 竞争失败的线程会通过循环一次次尝试进行 CAS操作，直到CAS 成功才会返回，也就是通过使用无限循环不断进行 CAS 尝试方式来替代阻塞算法挂起调用线程。相比阻塞算法，这是使用CPU资源换取阻塞所带来的开销。

add操作：

add操作是在链表尾部添加一个元素，其实在内部调用的还是offer操作。

public boolean add(E e) {
    return offer(e);
}

poll操作：

poll操作是在队列头部获取并移除一个元素，如果队列为空则返回null。

public E poll() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;

            if (item != null && p.casItem(item, null)) {
                // Successful CAS is the linearization point
                // for item to be removed from this queue.
                if (p != h) // hop two nodes at a time
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}

poll方法在移除一个元素时，只是简单地使用 CAS操作把当前节点的item值设置为null，然后通过重新设置头节点将该元素从队列里面移除，被移除的节点就成了孤立节点，这个节点会在垃圾回收时被回收掉。另外，如果在执行分支中发现头节点被修改了，要跳到外层循环重新获取新的头节点。

peak操作：

peak操作是获取队列头部获一个元素，如果队列为空则返回null。

public E peek() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;
            
            //注释
            if (item != null || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            }
            else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}

Peek操作的代码结构与poll操作类似，不同之处在于我们在代码中标记注释的地方中少了castItem操作。其实这很正常，因为peek只是获取队列头元素值，并不清空其值。根据前面的介绍我们知道第一次执行offer后head指向的是哨兵节点(也就是item为null的节点)，那么第一次执行peek时在注释处会发现item==null，然后执行q=p.next，这时候q节点指向的才是队列里面第一个真正的元素，或者如果队列为 null 则 q 指向 null。

加载全部内容