ConcurrentLinkedQueue 相关整理

1. ConcurrentLinkedQueue

• 在并发编程中有时候需要使用线程安全的队列，线程安全队列有两种实现方式。
  • 阻塞方式：对入队和出队操作加锁，阻塞队列。
  • 非阻塞方式：通过自旋 CAS 实现，例如：ConcurrentLinkedQueue。
    • 在中等规模的并发场景下，ConcurrentLinkedQueue 性能会高出不少，而且相当稳定。
• ConcurrentLinkedQueue 是一个基于 链接结点 的 无界线程安全队列。此队列按照 FIFO（先进先出）原则对元素进行排序。
  • 队列的头部是队列中时间最长的元素。
  • 队列的尾部是队列中时间最短的元素。
  • 新的元素插入到队列的尾部，队列获取操作从队列头部获得元素。
  • 当多个线程共享访问一个公共 collection 时，ConcurrentLinkedQueue 是一个恰当的选择。
  • 此队列不允许使用 null 元素。

1.1 ConcurrentLinkedQueue 的结构

 

ConcurrentLinkedQueue 的类图

• ConcurrentLinkedQueue 由 head 结点和 tail 结点组成，每个结点（Node）由结点元素（item）和指向下一个结点的引用 (next) 组成，结点与结点之间就是通过这个 next 关联起来，从而组成一张链表结构的队列。
• 默认情况下 head 结点存储的元素为空，tail 结点等于 head 结点。

        //Node代码中使用了UNSAFE提供的CAS方法保证操作的原子性，
        //UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val); 
        //第一个参数表示要更新的对象，第二个参数nextOffset是Field的偏移量，第三个参数表示期望值，最后一个参数更新后的值。若next域的值等于cmp，则把next域更新为val并返回true；否则不更新并返回false。
        private static class Node<E> {
            volatile E item;    //Node值，volatile保证可见性
            volatile Node<E> next;    //Node的下一个元素，volatile保证可见性

            /**
             * Constructs a new node.  Uses relaxed write because item can
             * only be seen after publication via casNext.
             */
            Node(E item) {
                UNSAFE.putObject(this, itemOffset, item);
            }

            boolean casItem(E cmp, E val) {
                return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
            }

            void lazySetNext(Node<E> val) {
                UNSAFE.putOrderedObject(this, nextOffset, val);
            }

            boolean casNext(Node<E> cmp, Node<E> val) {
                return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
            }

            // Unsafe mechanics

            private static final sun.misc.Unsafe UNSAFE;
            private static final long itemOffset;
            private static final long nextOffset;

            static {
                //初始化UNSAFE和各个域在类中的偏移量
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();//初始化UNSAFE
                    Class k = Node.class;
                    //itemOffset是指类中item字段在Node类中的偏移量，先通过反射获取类的item域，然后通过UNSAFE获取item域在内存中相对于Node类首地址的偏移量。
                    itemOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("item"));
                    //nextOffset是指类中next字段在Node类中的偏移量
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }
}

• Node 类中的 lazySetNext(Node<E> val) 方法，为延迟设置 next，内部使用 Unsafe 类的 putOrderedObject() 方法实现，putOrderedXXX 方法是 putXXXVolatile 方法的延迟实现，不保证值的改变被其他线程 立即看到。
  • 这种设置是不需要让共享变量的修改立刻让其他线程可见的时候，以设置普通变量的方式来修改共享状态，可以减少不必要的内存屏障，从而提高程序执行的效率。
    • volatile 变量也可以实现可见性，其原理是插入内存屏障以保证不会重排序指令，使用的是 StoreLoad 内存屏障，开销较大。
    • Unsafe 类的 putOrderedXXX 方法则是在指令中插入 StoreStore 内存屏障，只避免发生写操作重排序，由于 StoreStore 屏障的性能损耗小于 StoreLoad 屏障，所以 lazySetNext 方法比直接写 volatile 变量的性能要高。但注意的是，StoreStore 屏障仅可以避免写写重排序，不保证内存可见性。
• 初始化时候会构建一个 item 为 null 的空结点作为链表的首尾结点。

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

结点类型

• 有效结点：从 head 向后遍历可达的结点当中，item 域不为 null 的结点。
• 无效结点：从 head 向后遍历可达的结点当中，item 域为 null 的结点。
• 以删除结点：从 head 向后遍历不可达的结点。
• 哨兵结点：链接到自身的结点（哨兵结点同时也是以删除结点）。
• 头结点：队列中的第一个有效结点（如果有的话）。
• 尾结点：队列中 next 域为 null 的结点（可以是无效结点）。

1.2 入队操作

• 入队的方法为 offer，向队列的尾部插入指定的元素，由于 ConcurrentLinkedQueue 是无界的，所以 offer 永远返回 true，不能通过返回值来判断是否入队成功。
  • 入队源码中做了两件事情。
    • 定位出尾结点。
    • 使用 CAS 将入队结点设置成尾结点的 next 结点，如不成功则重试。

public boolean offer(E e) {
    // 如果e为null，则直接抛出NullPointerException异常
    checkNotNull(e);
    // 创建入队结点
    final Node<E> newNode = new Node<E>(e);
 
    // 循环CAS直到入队成功
    // 1、根据tail结点定位出尾结点（last node）；2、将新结点置为尾结点的下一个结点；3、casTail更新尾结点
    for (Node<E> t = tail, p = t;;) {
        // p用来表示队列的尾结点，初始情况下等于tail结点
        // q是p的next结点
        Node<E> q = p.next;
        // 判断p是不是尾结点，tail结点不一定是尾结点，判断是不是尾结点的依据是该结点的next是不是null
        // 如果p是尾结点
        if (q == null) {
            // p is last node
            // 设置p结点的下一个结点为新结点，设置成功则casNext返回true；否则返回false，说明有其他线程更新过尾结点
            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".
                // 如果p != t，则将入队结点设置成tail结点，更新失败了也没关系，因为失败了表示有其他线程成功更新了tail结点
                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
        }
        // 多线程操作时候，由于poll时候会把旧的head变为自引用，然后将head的next设置为新的head
        // 所以这里需要重新找新的head，因为新的head后面的结点才是激活的结点
        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;
    }
}

1.2.1 初始化状态

• 默认结点的 item 为 null，next 为 null。
• 构造函数中将默认结点赋值给 head 和 tail，head 等于 tail。
• 首次入队 q = p.next，因此 q 等于 null。

 

初始化状态

1.2.2 入队一个结点

• 因为 p 等于 t，p 就是尾结点，执行 p.casNext 通过 CAS 设置 p 的 next 为新增结点。

1.2.2.1 casNext 操作成功的情况

• 由于多线程可以调用 offer 方法，所以可能多个线程同时执行进行 CAS，那么只有一个会成功。
  • 假设当前线程 casNext 操作成功，退出循环。

 

casNext 设置成功状态

1.2.2.2 casNext 操作失败的情况

• 假设其他线程 p.casNext 设置成功，当前线程设置失败，则进入下一个循环。

 

casNext 设置失败的情况

• 这时 p 不等于 q，并且 t 等于 t，因此将 q 赋值给 p。
  • 因为 q = p.next，所以下一个循环中，q 再一次等于 null。

 

失败后根据条件设置

1.2.3 将入队结点设置为尾结点

• 假设当前线程 p.casNext 设置成功，其他线程造成 p 不等于 t（例如上述 casNext 操作失败的情况），那么当前线程会进行尾结点更新。

1.2.3.1 casTail 操作成功的情况

• 假设无干扰情况下，当前线程执行 casTail 操作成功，退出循环。

 

casTail 操作成功的情况

1.2.3.2 casTail 操作失败的情况

• 假设其他线程 casTail 设置成功，当前线程设置失败，则进入下一个循环。

 

casTail 操作失败的情况

• 这时 p 不等于 q，并且 t 不等于 newt（newt 等于 tail），因此将 newt 赋值给 p。
  • 这时的 p 等于 newt 等于 tail。
  • 因为 q = p.next，所以下一个循环中，q 再一次等于 null。

 

失败后根据条件设置

• 多个线程同时进行入队的情况，可能会出现其他线程插队的情况。
  • 如果有一个线程正在入队，那么它必须先获取尾结点，然后设置尾结点的下一个结点为入队结点，但这时可能有另外一个线程插队了，那么队列的尾结点就会发生变化，这时当前线程要暂停入队操作，然后重新获取尾结点。
• 入队操作主要做两件事情。
  • 第一是将入队结点设置成当前队列尾结点的下一个结点。
  • 第二是更新 tail 结点，如果 tail 结点的 next 结点不为空，则将入队结点设置成 tail 结点，如果 tail 结点的 next 结点为空，则将入队结点设置成 tail 的 next 结点，所以 tail 结点不总是尾结点。

1.2.3.3 p 等于 q 的情况处理

• 出队（poll）操作会造成 1.3.2.3 中的状态。

 

1.3.2.3 中的状态

• 这时 p 等于 q，将 tail 赋值给 t，t 与 t 相等，因此将 head 赋值给 p，进入下一个循环。

 

p 等于 q 的处理

• 入队新结点。
  • 因为 p 不等于 t，因此执行 casTail。

 

入队新结点

• 自引用的结点会被垃圾回收。

1.2.3.4 tail 结点不一定为尾结点的设计意图

• JDK 1.7
  • 使用 hops 变量来控制并减少 tail 结点的更新频率，并不是每次结点入队后都将 tail 结点更新成尾结点，而是当 tail 结点和尾结点的距离大于等于常量 HOPS 的值（默认等于 1）时才更新 tail 结点，tail 和尾结点的距离越长使用 CAS 更新 tail 结点的次数就会越少，但是距离越长带来的负面效果就是每次入队时定位尾结点的时间就越长，因为循环体需要多循环一次来定位出尾结点，但是这样仍然能提高入队的效率，因为从本质上来看它通过增加对 volatile 变量的读操作来减少了对 volatile 变量的写操作，而对 volatile 变量的写操作开销要远远大于读操作，所以入队效率会有所提升。
• JDK 1.8
  • tail 的更新时机是通过 p 和 t 是否相等来判断的，其实现结果和 JDK 1.7 相同，即当 tail 结点和尾结点的距离大于等于 1 时，更新 tail。

1.3 出队操作

• 出队的方法为 poll，出队就是从队列里返回一个结点元素，并清空该结点对元素的引用。
  • 并不是每次出队时都更新 head 结点，当 head 结点里有元素时，直接弹出 head 结点里的元素，而不会更新 head 结点。
  • 只有当 head 结点里没有元素时，出队操作才会更新 head 结点。
  • 采用这种方式也是为了减少使用 CAS 更新 head 结点的消耗，从而提高出队效率。

public E poll() {
    restartFromHead:
    for (;;) {
        // p结点表示首结点，即需要出队的结点
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;
 
            // 如果p结点的元素不为null，则通过CAS来设置p结点引用的元素为null，如果成功则返回p结点的元素
            if (item != null && p.casItem(item, null)) {
                // Successful CAS is the linearization point
                // for item to be removed from this queue.
                // 如果p != h，则更新head
                if (p != h) // hop two nodes at a time
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            // 如果头结点的元素为空或头结点发生了变化，这说明头结点已经被另外一个线程修改了。
            // 那么获取p结点的下一个结点，如果p结点的下一结点为null，则表明队列已经空了
            else if ((q = p.next) == null) {
                // 更新头结点
                updateHead(h, p);
                return null;
            }
            // p == q，则使用新的head重新开始
            else if (p == q)
                continue restartFromHead;
            // 如果下一个元素不为空，则将头结点的下一个结点设置成头结点
            else
                p = q;
        }
    }
}

• 该方法的主要逻辑是首先获取 head 结点的元素。
• 然后判断 head 结点元素是否为空，如果为空，表示另外一个线程已经进行了一次出队操作将该结点的元素取走，如果不为空，则使用 CAS 的方式将 head 结点的引用设置成 null。
• 如果 CAS 成功，则直接返回 head 结点的元素，如果不成功，表示另外一个线程已经进行了一次出队操作更新了 head 结点，导致元素发生了变化，需要重新获取 head 结点。

1.3.1 队列为空情况

1.3.1.1 执行过程中没有其他的线程入队结点

 

队列为空情况

• 因为 q = p.next 等于 null。

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

• 因为 h 等于 p，所以直接返回 null。

1.3.1.2 执行过程中其他线程入队了结点

 

执行过程中其他线程入队了结点

• 执行 q = p.next 前，其他线程已经入队了一个结点，q 不等于 null，p 不等于 q，因此将 q 赋值给 p，进入下一个循环。

 

执行过程中其他线程入队了结点

1.3.2 移除结点

1.3.2.1 casItem 操作成功至 updateHead 操作期间无其他操作情况

• 当 item 不等于 null，执行 casItem，只会有一个线程能执行成功，假设当前线程执行成功。
  • 因为 p 不等于 h，因此执行 updateHead 操作。
  • 又因为 q = p.next 等于 null，因此把 p 变为当前链表的 head 结点。
  • h 结点（旧的 head 结点）的 next 指向自己本身。

 

casItem 操作成功至 updateHead 操作期间无其他操作情况

1.3.2.2 casItem 操作失败的情况

• casItem 失败，说明有其他的线程执行成功。
  • 因为 q = p.next，所以 q 等于 null。
  • 执行 updateHead 操作，结果与之前一致。

 

casItem 操作失败的情况

1.3.2.3 casItem 操作成功至 updateHead 操作期间其他线程进行了出队

• 新线程出队操作进入设置 p = h。
• 之前的线程执行 updateHead 操作，q = p.next 不等于 null，因此把 q 变为当前链表的 head 结点。

 

casItem 操作成功至 updateHead 操作期间有其他线程进行了出队操作

• 因为 p 等于 q，因此通过 restartFromHead 跳出外层循环，重新查找 head 结点。

 

重新查找 head 结点

1.4 其他相关方法

1.4.1 peek 方法

• peek 操作是获取链表头部一个元素（只读取不移除）。
  • 代码与 poll 类似，只是少了 castItem。
  • peek 操作会改变 head 的指向。
    • 执行 peek 方法后 head 会指向第一个具有非空元素的结点。

// 获取链表的首部元素（只读取而不移除）
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;
        }
    }
}

1.4.2 size 方法

• size 方法用来获取当前队列的元素个数，但在并发环境中，其结果可能不精确，因为整个过程都没有加锁，所以从调用 size 方法到返回结果期间有可能增删元素，导致统计的元素个数不精确。

public int size() {
    int count = 0;
    // first() 获取第一个具有非空元素的结点，若不存在，返回 null
    // succ(p) 方法获取p的后继结点，若 p 等于 p.next，则返回 head
    for (Node<E> p = first(); p != null; p = succ(p))
        if (p.item != null)
            // Collection.size() spec says to max out
            // 最大返回Integer.MAX_VALUE
            if (++count == Integer.MAX_VALUE)
                break;
    return count;
}

//获取第一个具有非空元素的结点，没有则为 null
Node<E> first() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            boolean hasItem = (p.item != null);
            if (hasItem || (q = p.next) == null) {
                updateHead(h, p);
                return hasItem ? p : null;
            }
            else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}
 
//获取当前结点的 next 元素，如果是自引入结点则返回真正头结点
final Node<E> succ(Node<E> p) {
    Node<E> next = p.next;
    return (p == next) ? head : next;
}

1.4.3 remove 方法

• 如果队列里面存在该元素则删除该元素，如果存在多个则删除第一个，并返回 true，否者返回 false。

public boolean remove(Object o) {
    // 删除的元素不能为null
    if (o != null) {
        Node<E> next, pred = null;
 
        for (Node<E> p = first(); p != null; pred = p, p = next) {
            boolean removed = false;
            E item = p.item;

            // 结点元素不为null
            if (item != null) {
                // 若不匹配，则获取next结点继续匹配
                if (!o.equals(item)) {
                    next = succ(p);
                    continue;
                }

                // 若匹配，则通过CAS操作将对应结点元素置为null
                removed = p.casItem(item, null);
            }

            // 获取删除结点的后继结点
            next = succ(p);
            // 将被删除的结点移除队列
            if (pred != null && next != null) // unlink
                pred.casNext(p, next);
            if (removed)
                return true;
        }
    }
    return false;
}

1.4.4 contains 方法

• 判断队列里面是否含有指定对象，由于是遍历整个队列，所以类似 size 不是那么精确，有可能调用该方法时候元素还在队列里面，但是遍历过程中才把该元素删除了，那么就会返回 false。

public boolean contains(Object o) {
    if (o == null) return false;

    // 遍历队列
    for (Node<E> p = first(); p != null; p = succ(p)) {
        E item = p.item;
        // 若找到匹配结点，则返回true
        if (item != null && o.equals(item))
            return true;
    }
    return false;
}

1.4.5 add 方法

• 直接调用的 offer 方法。

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

1.5 总结

• 使用 CAS 原子指令来处理对数据的并发访问，这是非阻塞算法得以实现的基础。
• head/tail 并非总是指向队列的头 / 尾结点，也就是说允许队列处于不一致状态。
  • 这个特性把入队 / 出队时，原本需要一起原子化执行的两个步骤分离开来，从而缩小了入队 / 出队时需要原子化更新值的范围到唯一变量。这是非阻塞算法得以实现的关键。
• 由于队列有时会处于不一致状态。为此，ConcurrentLinkedQueue 使用 三个不变式 来维护非阻塞算法的正确性。

---

基本不变式

• 在执行方法之前和之后，队列必须要保持的不变式。
  • 当入队插入新结点之后，队列中有一个 next 为 null 的（最后）结点。
  • 从 head 开始遍历队列，可以访问所有 item 不为 null 的结点。

head 的不变式和可变式

• 在执行方法之前和之后，head 必须保持的不变式。
  • 所有 " 活着 " 的结点（指未删除结点），都能从 head 通过调用 succ() 方法遍历可达。
  • head 不能为 null。
  • head 结点的 next 不能引用到自身。
• 在执行方法之前和之后，head 的可变式。
  • head 结点的 item 可能为 null，也可能不为 null。
  • 允许 tail 滞后（lag behind）于 head，从 head 开始遍历队列，不一定能到达 tail。

tail 的不变式和可变式

• 在执行方法之前和之后，tail 必须保持的不变式。
  • 通过 tail 调用 succ() 方法，最后结点总是可达的。
  • tail 不能为 null。
• 在执行方法之前和之后，tail 的可变式。
  • tail 结点的 item 可能为 null，也可能不为 null。
  • 允许 tail 滞后于 head，从 head 开始遍历队列，不一定能到达 tail。
  • tail 结点的 next 可以引用到自身。

---

• 以批处理方式来更新 head/tail，从整体上减少入队 / 出队操作的开销。
• 为了有利于垃圾收集，队列使用特有的 head 更新机制。
  • 为了确保从已删除结点向后遍历，可到达所有的非删除结点，队列使用了特有的 向后推进 策略。

---

向后推进

• 由于 tail 可以指向任意结点，所以从 tail 向后遍历寻找尾结点的过程中，可能会遇到哨兵结点。此时 succ() 方法会直接跳转到 head 指向的结点继续遍历。
  • 如果 tail 滞后于 head，从 tail 向后遍历过程中就会发生跳转动作。
  • 跳转动作确保从已删除结点向后遍历，可以到达所有的未删除结点。

---

参考资料

https://blog.csdn.net/qq_38293564/article/details/80798310
http://www.importnew.com/25668.html
https://www.jianshu.com/p/7816c1361439

作者：羽杰
链接：https://www.jianshu.com/p/2b806ac8d28e
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。