有哪些并发队列?及ConcurrentLinkedQueue 源码分析

​

有哪些并发队列?说下某一个的实现？

ConcurrentLinkedQueue : 无界非阻塞队列，底层使用单向链表实现，对于出队和入队使用CAS来实现线程安全

LinkedBlockingQueue: 有界阻塞队列，使用单向链表实现，通过ReentrantLock实现线程安全，阻塞通过Condition实现，出队和入队各一把锁，不存在互相竞争

ArrayBlockingQueue: 有界数组方式实现的阻塞队列 , 通过ReentrantLock实现线程安全，阻塞通过Condition实现，出队和入队使用同一把锁

PriorityBlockingQueue: 带优先级的无界阻塞队列，内部使用平衡二叉树堆实现，遍历保证有序需要自定排序

DelayQueue: 无界阻塞延迟队列，队列中的每个元素都有个过期时间，当从队列获取元素时，只有过期元素才会出队列，队列头元素是最快要过期的元素
SynchronousQueue: 任何一个写需要等待一个读的操作，读操作也必须等待一个写操作，相当于数据交换
链接: 图解SynchronousQueue原理-公平模式 - BattleHeart - 博客园

LinkedTransferQueue: 由链表组成的无界阻塞队列，多了tryTransfer 和 transfer方法。transfer方法，能够把生产者元素立刻传输给消费者，如果没有消费者在等待，那就会放入队列的tail节点，并阻塞等待元素被消费了返回，可以使用带超时的方法。tryTransfer方法，会在没有消费者等待接收元素的时候马上返回false

LinkedBlockingDeque: 由链表组成的双向阻塞队列，可以从队列的两端插入和移除元素

​

源码基于Java8

无界非阻塞队列，底层使用单向链表实现，对于出队和入队使用CAS来实现线程安全。

看一下ConcurrentLinkedQueue的UML类图结构

AbstractQueue 实现了Queue接口的基本方法

Node 静态内部类，表示队列的节点

队列节点对象，ConcurrentLinkedQueue的静态内部类

private static class Node {
    //节点值
    volatile E item;
    //指向下一个节点
    volatile Node next;

    
    Node(E item) {
        UNSAFE.putObject(this, itemOffset, item);
    }

    //cas更新item值
    boolean casItem(E cmp, E val) {
        return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
    }

    //防重排序更新next值
    void lazySetNext(Node val) {
        UNSAFE.putOrderedObject(this, nextOffset, val);
    }

    //cas更新next值
    boolean casNext(Node cmp, Node val) {
        return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
    }

    private static final sun.misc.Unsafe UNSAFE;
    //item属性的偏移量
    private static final long itemOffset
    //next属性的偏移量
    private static final long nextOffset;

    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class k = Node.class;
            itemOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("item"));
            nextOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("next"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }
}

可拆分迭代器

static final class CLQSpliterator implements Spliterator {
    static final int MAX_BATCH = 1 << 25;  //最大批处理大小
    final ConcurrentLinkedQueue queue; 
    Node current;    //当前节点，初始化之前都为null
    int batch;          //分割批次的大小
    boolean exhausted;  //如果没有元素了，为true 
    CLQSpliterator(ConcurrentLinkedQueue queue) {
        this.queue = queue;
    }

    //分割队列,返回的是ArraySpliterator
    public Spliterator trySplit() {
        Node p;
        final ConcurrentLinkedQueue q = this.queue;
        int b = batch;
        //n表示数组长度
        //分割批次长度<=0，数组长度为1；
        //分割批次长度>=最大批处理大小，数组长度为最大批处理大小；否则为分割批次长度+1
        int n = (b <= 0) ? 1 : (b >= MAX_BATCH) ? MAX_BATCH : b + 1;
        //!exhausted 为true表示队列还有元素
        //((p = current) != null || (p = q.first()) != null) 如果前部分为true表示已经运行过一次，后部分为true表		  //示是第一次运行，将队列的头节点赋值给p
        //p.next != null 表示队列后面还有元素,p表示的节点不是最后一个
        if (!exhausted &&
            ((p = current) != null || (p = q.first()) != null) &&
            p.next != null) {
            //临时存储需分割的元素
            Object[] a = new Object[n];
            //已分割的元素计数，只计数item不为null的元素
            //对于ConcurrentLinkedQueue来说，这里不会有item值为null的情况
            int i = 0;
            do {
                //节点item值赋值临时数组
                if ((a[i] = p.item) != null)
                    //计数
                    ++i;
                //当next也指向本身p的时候为true,也就是自引用，重新找头节点
                if (p == (p = p.next))
                    p = q.first();
            } while (p != null && i < n);  //直到取完需要的元素数量
            //如果p等于null了就说明没有元素了
            if ((current = p) == null)
                exhausted = true;
            //有不为null的元素
            if (i > 0) {
                batch = i;
                //创建Spliterator，数组从0-i不包括i，指定了迭代器的特性
                return Spliterators.spliterator
                    (a, 0, i, Spliterator.ORDERED | Spliterator.NONNULL |
                     Spliterator.CONCURRENT);
            }
        }
        return null;
    }

    //循环每个节点并执行指定函数
    public void forEachRemaining(Consumer action) {
        Node p;
        if (action == null) throw new NullPointerException();
        final ConcurrentLinkedQueue q = this.queue;
        if (!exhausted &&
            ((p = current) != null || (p = q.first()) != null)) {
            exhausted = true;
            do {
                E e = p.item;
                if (p == (p = p.next))
                    p = q.first();
                if (e != null)
                    action.accept(e);
            } while (p != null); //循环每个节点，直到没有节点
        }
    }

    //获取一个有效的节点，执行指定函数
    public boolean tryAdvance(Consumer action) {
        Node p;
        if (action == null) throw new NullPointerException();
        final ConcurrentLinkedQueue q = this.queue;
        if (!exhausted &&
            ((p = current) != null || (p = q.first()) != null)) {
            E e;
            //通过循环的方式直到节点item值不为null
            //item值为null时通过p.next继续寻找
            do {
                e = p.item;
                //同样如果p是自引用就重新寻找头节点
                if (p == (p = p.next))
                    p = q.first();
            } while (e == null && p != null); 
            if ((current = p) == null)
                exhausted = true;
            if (e != null) {
                action.accept(e);
                return true;
            }
        }
        return false;
    }

    //返回估计的元素数量，因为ConcurrentLinkedQueue是无限队列，返回MAX_VALUE
    public long estimateSize() { return Long.MAX_VALUE; }

    //返回特征值
    public int characteristics() {
        return Spliterator.ORDERED | Spliterator.NONNULL |
            Spliterator.CONCURRENT;
    }
}

迭代器，从头节点到尾节点进行迭代，迭代器是弱一致性的，只是某一时刻的快照，因为同时有可能其他线程还在对队列进行修改

private class Itr implements Iterator {
    
    private Node nextNode;

    
    private E nextItem;

    
    private Node lastRet;

    Itr() {
        advance();
    }

    
    private E advance() {
        lastRet = nextNode;
        E x = nextItem;

        Node pred, p;
        //nextNode为null是说明是第一次运行
        if (nextNode == null) {
            //获取头节点
            p = first();
            pred = null;
        } else {
            //将上次获取的节点保存为此次的上一个节点
            pred = nextNode;
            //获取下一个节点
            p = succ(nextNode);
        }

        for (;;) {
            //p == null 的情况，可能是哨兵节点时，p=first()返回null
            //也可能是 p=succ(nextNode)到最后一个节点时，返回null
            if (p == null) {
                nextNode = null;
                nextItem = null;
                return x;
            }
            E item = p.item;
            if (item != null) {
                nextNode = p;
                nextItem = item;
                return x;
            } else {
                // 如果item=null时，即节点已被其他线程删除，那么继续获取下一个节点
                Node next = succ(p);
                if (pred != null && next != null)
                    //跳过null节点，将上一个节点的next指向新的节点
                    pred.casNext(p, next);
                	//将p赋值新节点，继续循环
                	p = next;
            }
        }
    }

    //通过下个项目节点不为null来判断是否还有下一个
    public boolean hasNext() {
        return nextNode != null;
    }

    //下一个项目
    public E next() {
        if (nextNode == null) throw new NoSuchElementException();
        return advance();
    }

    public void remove() {
        Node l = lastRet;
        if (l == null) throw new IllegalStateException();
        //将最近一个项目节点设置为null，在下一次执行advance时，该节点将被重新连接
        l.item = null;
        lastRet = null;
    }
}

先看下ConcurrentLinkedQueue有哪些基本属性

private transient volatile Node head;


private transient volatile Node tail;

Unsafe相关

private static final sun.misc.Unsafe UNSAFE;
//头节点相对于对象的偏移量
private static final long headOffset;
//尾节点相对于对象的偏移量
private static final long tailOffset;
static {
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class k = ConcurrentLinkedQueue.class;
        //使用Unsafe获取head属性相对于ConcurrentLinkedQueue对象的偏移量，可以使用该偏移量获取head属性的值
        headOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("head"));
        tailOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("tail"));
    } catch (Exception e) {
        throw new Error(e);
    }
}

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


public ConcurrentLinkedQueue(Collection c) {
    //定义头尾节点局部变量，h头节点变量，t尾节点变量
    Node h = null, t = null;
    for (E e : c) {
        //检查集合元素是否是null，否则将抛出NPE
        checkNotNull(e);
        //构造节点对象
        Node newNode = new Node(e);
        //如果t指向null，则h,t都指向第一个创建的node节点
        if (h == null)
            h = t = newNode;
        //否则说明已经存在头节点
        else {
            //①设置原node节点的next变量
            //lazySetNext使用了UNSAFE.putOrderedObject能够防止重排序
            t.lazySetNext(newNode);
            //②将t指向最后一个node 节点
            t = newNode;
        }
    }
    //当传入的集合为空集合时，局部变量h依然是null，将h,t指向item，next都为null的节点
    if (h == null)
        h = t = new Node(null);
    //更新全局头节点变量，尾节点变量
    head = h;
    tail = t;
}

无参构造函数创建对象后的节点图

根据给定的集合顺序创建队列对象，给定集合不是空集合

在尾部插入元素，实际调用offer()

在尾部插入元素

public boolean offer(E e) {
    //检查插入值不能为null
    checkNotNull(e);
    //构造node对象
    final Node newNode = new Node(e);

    for (Node t = tail, p = t;;) {
        Node q = p.next;
        //q->p.next,也就是说p已经是最后一个节点，p.next指向null
        if (q == null) {
            //cas设置，如果cas失败，说明有其他线程竞争，等待下次循环再次尝试cas设置
            if (p.casNext(null, newNode)) {
                //设置成功
                //尾节点与当前节点相隔两个，设置尾节点tail指向最后一个节点，即newNode
                if (p != t) 
                    //允许失败，下次会继续处理
                    casTail(t, newNode); 
                return true;
            }
        }
        //❓这里我在debug时有个奇怪问题
        //测试用例只有简单的add()方法，有时候能够进入该代码内，有时候执行的else部分代码
        //这个问题，见https://my.oschina.net/u/4311881/blog/3264783/print
        else if (p == q) 
            //这种情况会发生在节点已经被删除的时候，即p.next -> p 本身
            //如果tail没有发生变化，需要跳转到头节点head，从head节点能够到达任何存活节点
            //否则说明tail已经重新指向，用新的tail
            p = (t != (t = tail)) ? t : head;
        else
            //(p != t)  说明执行过 p = q 操作(向后遍历操作)
            //(t != (t = tail))) 说明尾节点在其他的线程发生变化
            //添加2次检查一次
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

public static void main(String[] args) {
    ConcurrentLinkedQueue