Java并发编程之ConcurrentLinkedQueue源码详解

// 链表头节点
private transient volatile node<E> head;
// 链表尾节点
private transient volatile Node<E> tail;

private static class Node<E> {
    volatile E item;
    volatile Node<E> next;
 
    
    Node(E item) {
        //将item存放在本节点的itemOffset偏移量位置的内存里
        UNSAFE.putObject(this, itemOffset, item);//设置this对象的itemoffset位置
    }
    //更新item值
    boolean casItem(E cmp, E val) {
       //this对象的itemoffset位置存放的值如果和期望值cmp相等，则替换为val
        return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
    }
 
    void lazySetNext(Node<E> val) {
      //this对象的nextOffset位置存入val
        UNSAFE.putOrderedObject(this, nextOffset, val);
    }
    //更新next节点值
    boolean casNext(Node<E> cmp, Node<E> val) {
     //this对象的nextOffset位置存放的值如果和期望值cmp相等，则替换为val
        return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
    }
 
    // Unsafe mechanics
 
    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 {
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class<?> k = ConcurrentLinkedQueue.class;
        headOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("head"));
        tailOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("tail"));
    } catch (Exception e) {
        throw new Error(e);
    }
}
 
private boolean casTail(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
}
 
private boolean casHead(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
}

//无参构造函数,head=tail=new Node<E>(null)=空节点
//初始一个为空的ConcurrentLinkedQueue,此时head和tail都指向一个item为null的节点
public ConcurrentLinkedQueue() {
    // 初始化头尾节点
    head = tail = new Node<E>(null);
}
 
//集合构造函数：就是将集合中的元素挨个链起来
public ConcurrentLinkedQueue(Collection<? extends E> c) {
    Node<E> h = null, t = null;
    for (E e : c) {
        checkNotNull(e);
        Node<E> newNode = new Node<E>(e);
        if (h == null)
            h = t = newNode;
        else {
            t.lazySetNext(newNode);//可以理解为一种懒加载,  将t的next值设置为newNode
            t = newNode;
        }
    }
    if (h == null)
        h = t = new Node<E>(null);
    head = h;
    tail = t;
}
 
private static void checkNotNull(Object v) {
    if (v == null)
        throw new NullPointerException();
}
 
//putObjectVolatile的内存非立即可见版本，
//写后结果并不会被其他线程看到,通常是几纳秒后被其他线程看到，这个时间比较短，所以代价可以接收
void lazySetNext(Node<E> val) {
    UNSAFE.putOrderedObject(this, nextOffset, val);
}

public boolean offer (E e){
    //先检查元素是否为null,是null则抛出异常 不是null,则构造新节点准备入队
    checkNotNull(e);
    final Node<E> newNode = new Node<E>(e);
    //初始p指针和t指针都指向尾节点，p指针用来向队列后面推移，t指针用来判断尾节点是否改变
    Node<E> t = tail, p = t;
    for (; ; ) {
        Node<E> q = p.next;
        if (q == null) {//p.next为null,则代表p为尾节点,则将p.next指向新节点 
            // p is last node
            if (p.casNext(null, newNode)) {
                
                if (p != t)
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
            // Lost CAS race to another thread; re-read next
        } else if (p == q)
        
 
            p = (t != (t = tail)) ? t : head;
        else
        
            // Check for tail updates after two hops.
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

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;
        }
    }
}