ConcurrentLinkedQueue
JDK中提供了一系列场景的并发安全队列。总的来说，按照实现方式的不同可分为阻塞队列和非阻塞队列，前者使用锁实现，而后则使用CAS非阻塞算法实现。
ConcurrentLinkedQueue 内部的队列使用单向链表方式实现，其中有两个volatile 类型的 Node 节点分别用来存放队列的首、尾节点。从下面的无参构造函数可知，默认头、尾节点都是指向 item 为null 的哨兵节点。新元素会被插入队列末尾，出队时从队列头部获取一个元素。
|
1
2
3
|
public ConcurrentLinkedQueue() { head = tail = new Node<E>(null);} |
在 Node 节点内部则维护一个使用volatile 修饰的变量 item，用来存放节点的值;next用来存放链表的下一个节点，从而链接为一个单向无界链表。其内部则使用 UNSafe 工具类提供的CAS 算法来保证出入队时操作链表的原子性。
下面通过介绍ConcurrentLinkedQueue的几个方法来介绍其实现原理。
offer操作： offer操作是在队列末尾添加一个元素，如果传递的参数是null则抛出NPE异常，否则由于ConcurrentLinkedQueue是无界队列，该方法一直会返回true。另外，由于使用CAS无阻塞算法，因此该方法不会阻塞挂起调用线程。下面具体看下实现原理。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
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操作。
|
1
2
3
|
public boolean add(E e) { return offer(e);} |
poll操作：
poll操作是在队列头部获取并移除一个元素，如果队列为空则返回null。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
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。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
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。
到此这篇关于Java并发编程之ConcurrentLinkedQueue队列详情的文章就介绍到这了,更多相关Java并发编程 ConcurrentLinkedQueue 内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://juejin.cn/post/7086080793478430750
