zoukankan html css js c++ java

Java同步数据结构之ConcurrentLinkedDeque

前言

由于LinkedBlockingDeque作为双端队列的实现，采用了单锁的保守策略使其不利于多线程并发情况下的使用，故ConcurrentLinkedDeque应运而生，它是一种基于链表的无界的同时支持FIFO、LIFO的非阻塞并发双端队列，当许多线程共享对公共集合的访问时，ConcurrentLinkedDeque是一个合适的选择，类比ConcurrentLinkedQueue是针对LinkedBlockingQueue对高并发情况的一种解决方案，ConcurrentLinkedDeque也是同样的地位，都是采用 CAS来替代加锁，甚至ConcurrentLinkedDeque再实现上也与ConcurrentLinkedQueue有很多相似的地方，其中最值得提及的就是，它采用了与ConcurrentLinkedQueue一样的松弛阀值设计（松弛阀值都是1），即head、tail并不总是指向队列的第一个、最后一个节点，而是保持head/tail距离第一个/最后一个节点的距离不超过1个节点的距离，从而减少了更新head/tail指针的CAS次数。Java Doc指出理解ConcurrentLinkedQueue的实现是理解该类实现的先决条件，所以最好先理解了ConcurrentLinkedQueue再来理解该类。

ConcurrentLinkedDeque另外还使用了两种方法来减少volatile写的次数：一是使用单次CAS操作来一次性使多次连续的CAS生效；二是将对同一块内存地址的volatile写与普通写混合。它的节点类与LinkedBlockingDeque的属性一致都是数据item、prev、next，只是多了一些CAS操作方法。与ConcurrentLinkedQueue一样，只有那些数据item不为空的节点才被认为是活动的节点，当将item置为null时，意味着从队列中逻辑删除掉了。

与LinkedBlockingDeque一样，任何时候，队列的第一个节点"first"的前驱prev为null，队列的最后一个节点"tail"的next后继为null。“first”和“last”节点可能是活动的，也可能不是活动的。“first”和“last”节点总是相互可达的。通过将第一个或最后一个节点的空前驱或后继CAS引用到包含指定元素的新节点，实现原子性地添加一个新元素，从而元素的节点在那时原子性地变成“活动的”，如果一个节点是活动的（item不为null）或者它是first/last节点，我们都称为是有效节点。ConcurrentLinkedDeque同样采用了“自链接（p.prev = p或p.next = p）”的方式使节点断开与队列的链接，有效活动节点不会有自链接的情况。

前面说了ConcurrentLinkedDeque有两个不总是指向第一个/最后一个节点的head、tail指针，所以它并没有像LinkedBlockingDeque那样设计first、tail属性，但是first、tail总是可以通过head、tail在O(1)时间内找到。

ConcurrentLinkedDeque删除节点分三个阶段：

logical deletion（逻辑删除）：通过CAS将数据item置为null，使该节点满足解除链接（unlinking）的条件。
unlinking（解除链接）：该阶段使队列中的活动节点无法到达该节点，但是保留该节点到队列中活动节点的链接，从而最终可由GC回收。此阶段典型的就是被迭代器使用的时候，使迭代器可以继续往下迭代。
gc-unlinking：该阶段进一步解除被删除节点到队列中活动节点的链接，使其更容易被GC回收，通过让节点自链接或链接到终止节点（PREV_TERMINATOR 或 NEXT_TERMINATOR）来实现。这一步是为了使数据结构保持GC健壮性(gc-robust)，消除使用保守式GC（conservative GC，目前已经很少使用）对内存无限期滞留的风险，并提高了分代GC的性能。

由于删除节点的第二、三阶段都不是保证数据正确性必须的，仅仅是对迭代器与内存的优化，故适当的减少这些操作的次数对性能是一种提高。所以ConcurrentLinkedDeque不仅设计了同ConcurrentLinkedQueue一样针对head、tail节点的松弛阈值，而且还提供了针对解除删除节点链接的阈值HOPS，也就是只有当逻辑删除的节点个数达到一定数量才会触发unlinking 和gc-unlinking，这样也是对性能的一种优化。

同ConcurrentLinkedQueue一样，ConcurrentLinkedDeque也对head、tail设定了如下的一些不变与可变性约束：

head/tail的不变性：

第一个节点总是可从head通过prev链接在O(1)时间复杂度内访问到。
最后一个节点总是可以从tail通过next链接在O(1)时间复杂度内访问到。
所有活动节点(item不为null)都可以从第一个节点通过succ()访问。
所有活动节点(item不为null)都可以从最后一个节点通过pred()访问。
head和tail都不会为null。
head节点的next不会指向自身形成自连接。
head/tail不会是GC-unlinked节点（但它可能是unlink节点）。

head/tail的可变性：

head、tail的数据item可以为null，也可以不为null。
head可能无法从第一个或最后一个节点或从tail到达。
tail可能无法从第一个或最后一个节点或从head到达。

下面开始分析ConcurrentLinkedDeque的源码，ConcurrentLinkedDeque和ConcurrentLinkedQueue并没有继承相应的BlockingQueue/BlockingQueue，容量又是无界的，所以不存在阻塞方法。

源码解析

首先看它的一些基础字段、节点内部类以及构造方法：

  1  /**
  2  * A node from which the first node on list (that is, the unique node p with p.prev == null && p.next != p) can be reached in O(1) time.
  3    可以在O（1）时间内从列表中的第一个节点到达的节点（即，具有p.prev == null && p.next！= p的唯一节点p）。
  4    
  5  * Invariants: 不变性
  6  * - the first node is always O(1) reachable from head via prev links 第一个节点总是可从head通过prev链接在O(1)时间内访问到
  7  * - all live nodes are reachable from the first node via succ() 所有活动节点都可以从第一个节点通过succ()访问
  8  * - head != null                                                  head不为空
  9  * - (tmp = head).next != tmp || tmp != head  
 10  * - head is never gc-unlinked (but may be unlinked) 。head永远不会gc-unlinked(但可能是unlinked)
 11  
 12  * Non-invariants: 可变性
 13  * - head.item may or may not be null                                           head 的数据项可以为空
 14  * - head may not be reachable from the first or last node, or from tail 。 head 可能无法从第一个或最后一个节点或从tail到达。
 15  */
 16 private transient volatile Node<E> head;
 17 
 18 /**
 19  * A node from which the last node on list (that is, the unique node p
 20  * with p.next == null && p.prev != p) can be reached in O(1) time.
 21    可以在O（1）时间内从列表中的最后一个节点到达的节点（即具有p.next == null && p.prev！= p的唯一节点p）。
 22 
 23  * Invariants: 不变性
 24  * - the last node is always O(1) reachable from tail via next links。最后一个节点始终可以通过下一个链接从tail访问在O(1)时间内访问到
 25  * - all live nodes are reachable from the last node via pred() 。所有活动节点都可以从最后一个节点通过pred()访问
 26  * - tail != null                                                            tail不为空
 27  * - tail is never gc-unlinked (but may be unlinked)                tail永远不会gc-unlinked(但可能是unlinked)
 28  
 29  * Non-invariants: 可变性
 30  * - tail.item may or may not be null                                            tail的数据项可以为空
 31  * - tail may not be reachable from the first or last node, or from head      tail可能无法从第一个或最后一个节点或从head访问到。
 32  */
 33 private transient volatile Node<E> tail;
 34 
 35 /**指示出队节点的终结节点*/
 36 private static final Node<Object> PREV_TERMINATOR, NEXT_TERMINATOR;
 37 
 38 
 39 @SuppressWarnings("unchecked")
 40 Node<E> prevTerminator() { //从对头出队节点的前向终结节点
 41     return (Node<E>) PREV_TERMINATOR;
 42 }
 43 
 44 @SuppressWarnings("unchecked")
 45 Node<E> nextTerminator() { //从对尾出队节点的后继终结节点
 46     return (Node<E>) NEXT_TERMINATOR;
 47 }
 48 
 49 static final class Node<E> {
 50     volatile Node<E> prev;
 51     volatile E item;
 52     volatile Node<E> next;
 53 
 54     Node() {  // default constructor for NEXT_TERMINATOR, PREV_TERMINATOR
 55     }
 56 
 57     /**
 58      * Constructs a new node.  Uses relaxed write because item can
 59      * only be seen after publication via casNext or casPrev.
 60      */
 61     Node(E item) {
 62         UNSAFE.putObject(this, itemOffset, item);
 63     }
 64 
 65     boolean casItem(E cmp, E val) {
 66         return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
 67     }
 68 
 69     void lazySetNext(Node<E> val) {
 70         UNSAFE.putOrderedObject(this, nextOffset, val);
 71     }
 72 
 73     boolean casNext(Node<E> cmp, Node<E> val) {
 74         return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
 75     }
 76 
 77     void lazySetPrev(Node<E> val) {
 78         UNSAFE.putOrderedObject(this, prevOffset, val);
 79     }
 80 
 81     boolean casPrev(Node<E> cmp, Node<E> val) {
 82         return UNSAFE.compareAndSwapObject(this, prevOffset, cmp, val);
 83     }
 84 
 85     // Unsafe mechanics
 86     private static final sun.misc.Unsafe UNSAFE;
 87     private static final long prevOffset;
 88     private static final long itemOffset;
 89     private static final long nextOffset;
 90 
 91     static {
 92         try {
 93             UNSAFE = sun.misc.Unsafe.getUnsafe();
 94             Class<?> k = Node.class;
 95             prevOffset = UNSAFE.objectFieldOffset
 96                 (k.getDeclaredField("prev"));
 97             itemOffset = UNSAFE.objectFieldOffset
 98                 (k.getDeclaredField("item"));
 99             nextOffset = UNSAFE.objectFieldOffset
100                 (k.getDeclaredField("next"));
101         } catch (Exception e) {
102             throw new Error(e);
103         }
104     }
105 }
106 
107 //针对被删除节点进行unlinking/GC-unlinking的阈值
108 private static final int HOPS = 2;
109 
110 private boolean casHead(Node<E> cmp, Node<E> val) {
111     return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
112 }
113 
114 private boolean casTail(Node<E> cmp, Node<E> val) {
115     return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
116 }
117 
118 // Unsafe mechanics
119 private static final sun.misc.Unsafe UNSAFE;
120 private static final long headOffset;
121 private static final long tailOffset;
122 static {
123     PREV_TERMINATOR = new Node<Object>();
124     PREV_TERMINATOR.next = PREV_TERMINATOR;
125     NEXT_TERMINATOR = new Node<Object>();
126     NEXT_TERMINATOR.prev = NEXT_TERMINATOR;
127     try {
128         UNSAFE = sun.misc.Unsafe.getUnsafe();
129         Class<?> k = ConcurrentLinkedDeque.class;
130         headOffset = UNSAFE.objectFieldOffset
131             (k.getDeclaredField("head"));
132         tailOffset = UNSAFE.objectFieldOffset
133             (k.getDeclaredField("tail"));
134     } catch (Exception e) {
135         throw new Error(e);
136     }
137 }
138 
139 /**
140  * Constructs an empty deque. 默认构造方法，head、tail都指向同一个item为null的节点
141  */
142 public ConcurrentLinkedDeque() {
143     head = tail = new Node<E>(null);
144 }
145 
146 /**
147  * Constructs a deque initially containing the elements of
148  * the given collection, added in traversal order of the
149  * collection's iterator.
150  *
151  * @param c the collection of elements to initially contain
152  * @throws NullPointerException if the specified collection or any
153  *         of its elements are null
154  */
155 public ConcurrentLinkedDeque(Collection<? extends E> c) {
156     // Copy c into a private chain of Nodes
157     Node<E> h = null, t = null;
158     for (E e : c) {
159         checkNotNull(e);
160         Node<E> newNode = new Node<E>(e);
161         if (h == null)
162             h = t = newNode;
163         else {
164             t.lazySetNext(newNode);
165             newNode.lazySetPrev(t);
166             t = newNode;
167         }
168     }
169     initHeadTail(h, t);
170 }
171 
172 /**
173  * Initializes head and tail, ensuring invariants hold.
174  * 初始化head和tail，确保它们的不变性
175  */
176 private void initHeadTail(Node<E> h, Node<E> t) {
177     if (h == t) { //队列为空，或者只有一个元素
178         if (h == null)
179             h = t = new Node<E>(null);//队列为空，head、tail都指向同一个item为null的节点
180         else { 
181             // 只有一个元素，重新构造一个节点指向tail，避免head、tail都指向同一个非null节点
182             // Avoid edge case of a single Node with non-null item.
183             Node<E> newNode = new Node<E>(null);
184             t.lazySetNext(newNode);
185             newNode.lazySetPrev(t);
186             t = newNode;
187         }
188     }
189     head = h;
190     tail = t;
191 }

View Code

节点内部类和LinkedBlockingDeque一样都是prev、tail、item，空队列情况下，head、tail都指向一个item为null的节点。PREV_TERMINATOR、NEXT_TERMINATOR分别是从对头/队尾出队节点的前向/后继终止节点。ConcurrentLinkedDeque是无界的。

入队实现

头部入队

 1 /**
 2  * Links e as first element. 在头节点入队
 3  */
 4 private void linkFirst(E e) {
 5     checkNotNull(e);
 6     final Node<E> newNode = new Node<E>(e);
 7 
 8     restartFromHead:
 9     for (;;)
10         //从head节点往前（左）寻找first节点
11         for (Node<E> h = head, p = h, q;;) {
12             if ((q = p.prev) != null &&     //前驱不为null
13                 (q = (p = q).prev) != null) //前驱的前驱也不为null(有线程刚刚从对头入队了一个节点，还没来得及修改head)
14                 // Check for head updates every other hop.
15                 // If p == q, we are sure to follow head instead.
16                 p = (h != (h = head)) ? h : q; //head被更新了就重新取head，否则取前驱的前驱
17             else if (p.next == p) // PREV_TERMINATOR   p是第一个节点，但是是自链接，表示出队了，重新开始
18                 continue restartFromHead;
19             else {
20                 // p是第一个节点
21                 newNode.lazySetNext(p); // p成为新节点的后继节点
22                 if (p.casPrev(null, newNode)) { //新节点成为p的前驱节点
23                     //成功将e入队
24                     if (p != h) // 松弛阀值超过1，更新head
25                         casHead(h, newNode);  // Failure is OK.
26                     return;
27                 }
28                 // 失败，可能被其它线程抢先入队，重新找前驱
29             }
30         }
31 }

同LinkedBlockingDeque一样，linkFirst是从对头入队新节点的具体逻辑实现（被其它入队方法调用），看起来很简单：从head节点往对头寻找第一个节点p（不论item是不是null），找到之后将新节点链接到它的前驱，同时当head的松弛阈值超过1时更新head。linkFirst分别被offerFirst、addFirst、push方法直接或间接调用。

尾部入队

 1 /**
 2  * Links e as last element. 在队尾入队
 3  * tail -> A 
 4  */
 5 private void linkLast(E e) {
 6     checkNotNull(e);
 7     final Node<E> newNode = new Node<E>(e);
 8 
 9     restartFromTail:
10     for (;;)
11     //从tail向后（右）找最后一个节点
12         for (Node<E> t = tail, p = t, q;;) {
13             if ((q = p.next) != null &&      //p的后继不为null
14                 (q = (p = q).next) != null)  //后继的后继也不为null(有线程刚刚入队了一个节点，还没来得及修改tail)
15                 // Check for tail updates every other hop.
16                 // If p == q, we are sure to follow tail instead.
17                 p = (t != (t = tail)) ? t : q;  //若tail被更新了就重新从tail开始往后查找，否则取后继的后继
18             else if (p.prev == p) // NEXT_TERMINATOR   p是最后一个节点，但是是自链接了，表示出队了，重新开始
19                 continue restartFromTail;
20             else {
21                 // p 是最后一个节点
22                 newNode.lazySetPrev(p); // p成为新节点的前驱
23                 if (p.casNext(null, newNode)) { //新节点成为p的后继
24                     //成功将e入队
25                     if (p != t) // 松弛阀值超过1，更新tail
26                         casTail(t, newNode);  // Failure is OK.
27                     return;
28                 }
29                 // 失败，可能被其它线程抢先入队，重新找后继节点
30             }
31         }
32 }

View Code

队尾入队的逻辑基本上和linkFirst一样，不同的是它是从tail节点往后寻找最后一个节点，把新节点链接到它的后继，同时维护tail的松弛阈值。linkLast分别被offerLast、addLast、add、offer方法直接或间接调用。

入队的逻辑流程图如下（ABC分别从队尾入队，DE从对头入队）；

出队

  1 //对头出队
  2 public E pollFirst() {
  3     for (Node<E> p = first(); p != null; p = succ(p)) {
  4         E item = p.item;
  5         //只对item不为null的节点做处理，先将item置为null（逻辑删除），再进行断开连接操作
  6         if (item != null && p.casItem(item, null)) {
  7             unlink(p);
  8             return item;
  9         }
 10     }
 11     return null;
 12 }
 13 
 14 /**
 15  * Unlinks non-null node x. 将非空节点x出队
 16  */
 17 void unlink(Node<E> x) {
 18     // assert x != null;
 19     // assert x.item == null;
 20     // assert x != PREV_TERMINATOR;
 21     // assert x != NEXT_TERMINATOR;
 22 
 23     final Node<E> prev = x.prev;
 24     final Node<E> next = x.next;
 25     if (prev == null) { //前驱为空，表示是第一个节点
 26         unlinkFirst(x, next);
 27     } else if (next == null) {//后继为空，表示是最后一个节点
 28         unlinkLast(x, prev);
 29     } else {
 30         //内部节点出队
 31         //
 32         // This is the common case, since a series of polls at the same end will be "interior" removes, except perhaps for the first one, since end nodes cannot be unlinked.
 33         // 这是常见的情况，因为在同一端的一系列poll都将被“内部”删除，除了第一个，因为末端节点不能被取消链接。
 34         
 35         // At any time, all active nodes are mutually reachable by following a sequence of either next or prev pointers.
 36         // 在任何时候，所有活动节点都可以通过循着next或prev指针相互访问。
 37         
 38         // Our strategy is to find the unique active predecessor and successor of x.  Try to fix up their links so that they point to each other, leaving x unreachable from active nodes.  我们的策略是找到x唯一的活动前驱和后继节点。并尝试修改它们的链接，使它们彼此指向对方，使x节点无法从活动节点被访问。
 39         If successful, and if x has no live predecessor/successor, we additionally try to gc-unlink, leaving active nodes unreachable from x, by rechecking that the status of predecessor and successor are unchanged and ensuring that x is not reachable from tail/head, before setting x's prev/next links to their logical approximate replacements, self/TERMINATOR.
 40         如果成功,并且如果x没有活动前驱/后继,我们进行gc-unlink,使x无法访问到活动节点,通过查看前驱和后继的状态没有被改变，确保x节点不能从head/tail节点被访问，
 41         之前设置x前驱/后继的链接到它们的逻辑近似替代,自身/终结者。
 42         
 43         Node<E> activePred, activeSucc;
 44         boolean isFirst, isLast;
 45         int hops = 1;
 46 
 47         // 从被删除节点往前找到第一个有效前驱节点
 48         for (Node<E> p = prev; ; ++hops) {
 49             if (p.item != null) { //找到有效节点
 50                 activePred = p;
 51                 isFirst = false;
 52                 break;
 53             }
 54             Node<E> q = p.prev;
 55             if (q == null) { //已经到对头了
 56                 if (p.next == p) //发现自连接，直接返回
 57                     return;
 58                 activePred = p;
 59                 isFirst = true;
 60                 break;
 61             }
 62             else if (p == q)//发现自连接，直接返回
 63                 return;
 64             else //更新循环指针
 65                 p = q;
 66         }
 67 
 68         // 从被删除节点往后找到第一个有效后继节点
 69         for (Node<E> p = next; ; ++hops) {
 70             if (p.item != null) {//找到有效节点
 71                 activeSucc = p;
 72                 isLast = false;
 73                 break;
 74             }
 75             Node<E> q = p.next;
 76             if (q == null) {//已经到对尾了
 77                 if (p.prev == p) //发现自连接，直接返回
 78                     return;
 79                 activeSucc = p;
 80                 isLast = true;
 81                 break;
 82             }//发现自连接，直接返回
 83             else if (p == q)
 84                 return;
 85             else
 86                 p = q; //更新循环指针
 87         }
 88 
 89         // TODO: better HOP heuristics
 90         //如果已经积累了超过阈值的逻辑删除节点，或者是内部节点删除，我们需要进一步处理unlink/gc-unlink
 91         if (hops < HOPS
 92             // always squeeze out interior deleted nodes
 93             && (isFirst | isLast))
 94             return;
 95 
 96         // Squeeze out deleted nodes between activePred and activeSucc, including x.
 97         // 移除有效前驱和后继节点之间的那些节点（都是逻辑删除的节点），包括x节点本身
 98         // 就是使有效前驱好后继节点互连（unlink）
 99         skipDeletedSuccessors(activePred);
100         skipDeletedPredecessors(activeSucc);
101 
102         // Try to gc-unlink, if possible 
103         //如果更新的开头或者结尾，那么就可以尝试进行gc-unlink
104         if ((isFirst | isLast) &&
105             //确保前驱和后继的状态没有被改变
106             // Recheck expected state of predecessor and successor
107             (activePred.next == activeSucc) &&
108             (activeSucc.prev == activePred) &&
109             (isFirst ? activePred.prev == null : activePred.item != null) &&
110             (isLast  ? activeSucc.next == null : activeSucc.item != null)) {
111 
112             //确保x节点不能从head/tail节点被访问，
113             updateHead(); // Ensure x is not reachable from head
114             updateTail(); // Ensure x is not reachable from tail
115 
116             // Finally, actually gc-unlink
117             x.lazySetPrev(isFirst ? prevTerminator() : x);//前向终结节点
118             x.lazySetNext(isLast  ? nextTerminator() : x);//后继终结节点
119         }
120     }
121 }
122  
123 /**
124  * Unlinks non-null first node. 从对头将第一个非空节点出队
125  * first就是第一个item不为null的节点，next是该first节点的后继节点，此方法只被unlink调用
126  * 就是从first开始往后找到第一个有效节点，直到找到或者到达队列的最后一个节点为止，并把first的直接后继指向该有效节点：
127  * 1）如果first的原后继本身就是有效节点，不做任何处理；
128  * 2）否则往后依次找到第一个有效节点，并把first的后继指向该有效节点。
129  */
130 private void unlinkFirst(Node<E> first, Node<E> next) {
131     // assert first != null;
132     // assert next != null;
133     // assert first.item == null;
134     for (Node<E> o = null, p = next, q;;) {
135         //p是有效节点，或者p是队列中的最后一个节点
136         if (p.item != null || (q = p.next) == null) {
137             
138             //第一次for循环时o为null，p是first的有效后继节点，否则o是一个item为null的节点，p是o的有效后继节点
139             
140             
141             //超过一次for循环时o不为null，p的o的后继有效节点，并且p还没有断开还指向o，将first后继指向该有效节点p
142             //也就是说，p是first后面的第N个后继，N >= 2
143             if (o != null && p.prev != p && first.casNext(next, p)) {
144                 skipDeletedPredecessors(p); （unlink）
145                 
146                 //下面这个if其实就是在确保first还是第一个节点，没有被其他线程改变状态
147                 //并且它和它的后继节点p是直接相连的，这种关系没有被破坏，进行gc-unlink
148                 if (first.prev == null &&  
149                     (p.next == null || p.item != null) &&
150                     p.prev == first) {
151 
152                     updateHead(); // Ensure o is not reachable from head
153                     updateTail(); // Ensure o is not reachable from tail
154 
155                     // 最终的gc-unlink（解除被删除节点到活动节点的链接）
156                     o.lazySetNext(o); //o后继自连接
157                     o.lazySetPrev(prevTerminator()); //o前驱指向PREV_TERMINATOR
158                 }
159             }
160             //要么p本来就是first的有效后继，要么已经通过上面first.casNext(next, p)将有效节点p设置成first的后继节点了
161             return;
162         }
163         //p不是有效节点，并且p已经与后继节点断开（自链接），直接返回
164         //既然first后面已经有节点断开连接，所以不需要做多余处理
165         else if (p == q)
166             return;
167         else { //p不是有效节点，但p还有后继节点q，循环指针指向q
168             o = p;
169             p = q;
170         }
171     }
172 }
173 
174 //跳过已经删除的前驱节点
175 //其实就是从x开始往（左）依次找到第一个有效节点，直至找到或者到达队列的对头，
176 //然后将该节点设置成x的前驱
177 private void skipDeletedPredecessors(Node<E> x) {
178     whileActive:
179     do {
180         Node<E> prev = x.prev;
181         // assert prev != null;
182         // assert x != NEXT_TERMINATOR;
183         // assert x != PREV_TERMINATOR;
184         Node<E> p = prev;
185         findActive:
186         for (;;) {
187             if (p.item != null) //找到有效前驱p
188                 break findActive;
189             Node<E> q = p.prev;
190             if (q == null) { //p已经是第一个节点
191                 if (p.next == p)//发现自连接，重新开始
192                     continue whileActive;
193                 break findActive;
194             }
195             else if (p == q) //发现自连接，重新开始
196                 continue whileActive;
197             else
198                 p = q; //循环指针指向前驱
199         }
200 
201         // 如果找到的有效前驱本身就是x的直接前驱，不做处理直接返回，否则将x的前驱指向该新的前驱
202         if (prev == p || x.casPrev(prev, p))
203             return;
204 
205     } while (x.item != null || x.next == null);//x是有效节点，或是队列最后一个节点
206 }
207 
208 /**
209  * Guarantees that any node which was unlinked before a call to this method will be unreachable from head after it returns.
210  * 确保在调用此方法之前unlinked的任何节点在该方法返回之后都不能从head访问。
211  * Does not guarantee to eliminate slack, only that head will point to a node that was active while this method was running.
212  * 不保证消除松弛，仅仅是head将指向在此方法运行时处于活动状态的节点。
213  */
214 private final void updateHead() {
215     // Either head already points to an active node, or we keep trying to cas it to the first node until it does.
216     // 要么head已经指向一个活动节点，要么我们一直尝试将head指向第一个有效节点，直到成功。
217     Node<E> h, p, q;
218     restartFromHead:
219     while ((h = head).item == null && (p = h.prev) != null) {//head并不是指向有效节点，也不是指向第一个节点
220         for (;;) {
221             if ((q = p.prev) == null || //head前驱p的前驱q为null 或者
222                 (q = (p = q).prev) == null) {//head前驱的前驱p的前驱q为null
223                 // It is possible that p is PREV_TERMINATOR,
224                 // but if so, the CAS is guaranteed to fail.
225                 //说明p（head的前驱或前驱的前驱）是第一个节点，将head指向第一个节点
226                 if (casHead(h, p))
227                     return;
228                 else //失败重新开始
229                     continue restartFromHead;
230             }
231             //说明head与第一个节点的距离超过2，如果head被改变重新开始
232             else if (h != head)
233                 continue restartFromHead;
234                 
235                 //  q <- p1 <- p0 <- head
236             else ////说明head与第一个节点的距离超过2，head还没被改变，循环指针向前跳到其前驱的前驱的前驱
237                 p = q;
238         }
239     }
240 }

View Code

这里以pollFirst出队方法为例，其他方法逻辑都一样，先通过first()拿到队列头部的第一个节点，如果是活动节点（item不为null），则直接将item置为null，即完成了删除节点的第一步逻辑删除，然后执行unlink方法执行删除节点的第二unlinking、第三步GC-unlinking，unlink方法针对节点在不同的位置按不同的逻辑处理，①如果出队的节点是队列的第一个节点，则执行unlinkFirst；②如果是队列的最后一个节点，则执行unlinkLast，③否则表示是内部节点，执行unlink本身的通用节点逻辑。

unlinkFirst的逻辑其实就分两个部分：①实现从被移除节点p开始往后（队尾）找到第一个有效节点，直到找到或者到达队列的最后一个节点为止，并把p的直接后继指向该有效节点（如果本身不是其后继节点的话），其中的skipDeletedPredecessors方法实现将刚刚找到的后继节点的前驱也指向节点p，即完成它们的互联，这一步就是所谓的unlinking，使队列的活动节点无法访问被删除的节点；②第二部分就是实现GC-unlinking了，通过updateHead、updateTail使被删除的节点无法从head/tail可达，最后让被删除节点后继自连接，前驱指向前向终结节点。

如果是内部节点出队，执行unlink本身：先找到被删除节点x的有效前驱和后继节点，并记录它们中间的已经被逻辑删除的节点个数，如果已经积累了超过阈值的节点个数，或者是内部节点删除，我们需要进一步处理unlink/gc-unlink，①首先使被删除节点的有效前驱节点和后继节点互联，就相当于导致活动节点不会访问到中间已经被逻辑删除的节点（unlinking）；②若第①步导致重新链接到了对头或队尾，则通过updateHead、updateTail使被删除的节点无法从head/tail可达，最后让被删除节点自连接或者执行终结节点（GC-unlinking）。

如果是队尾节点出队则由unlinkLast，unlinkLast的源码其实与unlinkFirst基本一致，只不过是从被删除节点p往前寻找一个有效节点，并把p的直接前驱节点指向该有效节点（如果本身不是其前驱节点的话），其中skipDeletedSuccessors则让刚刚找到的前驱节点的后继也指向节点p，即完成它们的互联，这一步就是所谓的unlinking，使队列的活动节点无法访问被删除的节点；②第二部分就是实现GC-unlinking了，通过updateHead、updateTail使被删除的节点无法从head/tail可达，最后让被删除节点前驱自连接，后继指向后继终结节点。unlinkLast的源码就不贴了。

可以看见，ConcurrentLinkedDeque在实现的时候，其实对头队尾相关的方法都是对称的，所以理解了一端的方法，另一端的方法就是对称的。

出队的方法主要就是unlink + unlinkFirst + unlinkLast实现，它被ConcurrentLinkedDeque的其他方法调用，例如：pollFirst、removeFirst、remove(包括迭代器)、clear、poll、pollLast、removeLast、removeFirstOccurrence(Object o)、removeLastOccurrence(Object o)等大量方法直接或间接调用。

其它方法

peekFirst/peekLast方法从对头/队尾开始找第一个活动节点（item不为空），找到一个立即返回item数据，否则直到到达队列的另一端都没找到返回null。这两个方法分别还会被peek/getFirst/isEmpty/getLast方法调用。例如isEmpty方法调用peekFirst只要返回不为null就表示队列非空，

size() ，返回当前时刻队列中item不为空的节点个数，但如果超过Integer.MAX_VALUE，则就返回Integer.MAX_VALUE。

addAll(Collection c), 将指定的集合组成一个临时双端队列，然后把该临时队列拼接到当前ConcurrentLinkedDeque队列的队尾。指定的参数集合不能是ConcurrentLinkedDeque本身，不然将抛出IllegalArgumentException异常。

toArray/toArray(T[] a)，从队头开始依次将item不为空的节点数据添加到一个ArrayList集合中，最后再通过toArray方法将其转换成数组，注意该方法并不会将数据从队列中移除，仅仅是拷贝item的引用，所以返回的数组可以任意操作而不会对队列本身造成任何影响。

迭代器

ConcurrentLinkedDeque的迭代器实现思想与LinkedBlockingDeque一致，也支持正向和逆向的两种迭代器，分别是方法iterator、descendingIterator：

 1 //按正确的顺序返回deque中元素的迭代器。元素将按从第一个(head)到最后一个(tail)的顺序返回。
 2 //返回的迭代器是弱一致的。
 3 public Iterator<E> iterator() {
 4     return new Itr();
 5 }
 6     
 7 //以相反的顺序返回deque中元素的迭代器。元素将按从最后(tail)到第一个(head)的顺序返回。
 8 //返回的迭代器是弱一致的。
 9 public Iterator<E> descendingIterator() {
10     return new DescendingItr();
11 }

它们的逻辑主要是由一个内部抽象类AbstractItr来实现，而iterator和descendingIterator仅仅实现了AbstractItr的抽象方法，用来指示迭代器的开始位置和迭代方向，为了保证迭代器的弱一致性，迭代器在创建实例的时候就已经拿到了第一个节点next和其节点数据，为了实现迭代器的remove方法，迭代器还保留了迭代的上一个节点lastRet，用于获取迭代器的下一个节点的主要逻辑由advance方法实现：

 1 /**
 2  * Sets nextNode and nextItem to next valid node, or to null if no such.
 3  * 将nextNode和nextItem设置为下一个有效节点，如果没有，则设置为null。
 4  */
 5 private void advance() {
 6     lastRet = nextNode;
 7 
 8     Node<E> p = (nextNode == null) ? startNode() : nextNode(nextNode);
 9     for (;; p = nextNode(p)) {
10         if (p == null) { //队列没有更多元素了，空或者到头了
11             // p might be active end or TERMINATOR node; both are OK
12             nextNode = null;
13             nextItem = null;
14             break;
15         }
16         E item = p.item;
17         if (item != null) {//节点有效
18             nextNode = p;
19             nextItem = item;
20             break;
21         }
22     }
23 }

View Code

可见迭代器会排除那些被移除的无效节点，迭代器在使用Itr.remove()删除节点的时候实际上调用了ConcurrentLinkedDeque的unlink方法，该方法上面已经解析过了，其它方法都很简单就不一一列举了。

可拆分迭代器Spliterator

ConcurrentLinkedDeque的可拆分迭代器由内部类CLDSpliterator实现，它不像普通迭代器那样可以支持正向和反向迭代，可拆分迭代器仅支持正向的拆分迭代：

1 public Spliterator<E> spliterator() {
2       return new CLDSpliterator<E>(this);
3 }

ConcurrentLinkedDeque的可拆分迭代器实现基本上和LinkedBlockingDeque一样，不过它不是使用锁而是CAS实现，可拆分迭代器会对节点的数据item进行null值判断，只对item不为空的数据做处理，tryAdvance从对头开始查找获取队列中第一个item不为空的数据节点的数据做指定的操作，forEachRemaining从队头开始循环遍历当前队列中item不为空的数据节点的数据做指定的操作源码都很简单，就不贴代码了，至于它的拆分方法trySplit，其实和ConcurrentLinkedQueue/LinkedBlockingDeque拆分方式是一样的，代码都几乎一致，它不是像ArrayBlockingQueue那样每次分一半，而是第一次只拆一个元素，第二次拆2个，第三次拆三个，依次内推，拆分的次数越多，拆分出的新迭代器分的得元素越多，直到一个很大的数MAX_BATCH（33554432），后面的迭代器每次都分到这么多的元素，拆分的实现逻辑很简单，每一次拆分结束都记录下拆分到哪个元素，下一次拆分从上次结束的位置继续往下拆分，直到没有元素可拆分了返回null。

总结

ConcurrentLinkedDeque是双端队列家族中对LinkedBlockingDeque的一种高并发优化，因为LinkedBlockingDeque采用的是保守的单锁实现，在多线程高并发下效率极其低下，所以ConcurrentLinkedDeque采用了CAS的方法来处理所以的竞争问题，保留了双端队列的所有特性，可以从对头、对尾两端插入和移除元素，它的内部实现非常精妙，既采用了ConcurrentLinkedQueue实现中用到过松弛阈值处理（即并不每一次都更新head/tail指针），又独特的针对队列中被逻辑删除节点的进行了淤积阀值合并处理和分三个阶段的节点删除步骤，同时还针对多次volatile写、普通写，多次连续的CAS操作单次生效等一系列的措施减少volatile写和CAS的次数，提高ConcurrentLinkedDeque的运行效率。当许多线程共享对公共集合（双端队列）的访问时，ConcurrentLinkedDeque是一个合适的选择，如果不需要用到双端队列的特性，完全可以使用ConcurrentLinkedQueue来完成高并发对公共集合的高效使用。注意ConcurrentLinkedDeque，ConcurrentLinkedQueue都没有继承BlockingDeque、BlockingQueue，所以它们没有阻塞等待的相关方法。

查看全文

相关阅读:
10个实用的但偏执的Java编程技术
 分布式文件系统FastDFS设计原理
 👮 Golang Gin/Ace/Iris/Echo RBAC 鉴权库
 深入浅出爬虫之道： Python、Golang与GraphQuery的对比
 (.*?)
开箱即用的高匿代理抓取工具
 6.可见性关键字（volidate）
5.同步关键字（synchronized）
4.等待和通知
 3.线程优先级

原文地址：https://www.cnblogs.com/txmfz/p/10975750.html