private static class Node<E> {// 元素volatile E item;// next域volatile Node<E> next;/*** Constructs a new node.  Uses relaxed write because item can* only be seen after publication via casNext.*/// 构造函数
        Node(E item) {// 设置item的值UNSAFE.putObject(this, itemOffset, item);}// 比较并替换item值boolean casItem(E cmp, E val) {return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);}void lazySetNext(Node<E> val) {// 设置next域的值，并不会保证修改对其他线程立即可见UNSAFE.putOrderedObject(this, nextOffset, val);}// 比较并替换next域的值boolean casNext(Node<E> cmp, Node<E> 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);}}}

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>implements Queue<E>, java.io.Serializable {// 版本序列号        private static final long serialVersionUID = 196745693267521676L;// 反射机制private static final sun.misc.Unsafe UNSAFE;// head域的偏移量private static final long headOffset;// tail域的偏移量private static final long tailOffset;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 transient volatile Node<E> head;// 尾结点private transient volatile Node<E> tail;
}

    public ConcurrentLinkedQueue() {// 初始化头结点与尾结点head = tail = new Node<E>(null);}

    public ConcurrentLinkedQueue(Collection<? extends E> c) {Node<E> h = null, t = null;for (E e : c) { // 遍历c集合// 保证元素不为空
            checkNotNull(e);// 新生一个结点Node<E> newNode = new Node<E>(e);if (h == null) // 头结点为null// 赋值头结点与尾结点h = t = newNode;else {// 直接头结点的next域
                t.lazySetNext(newNode);// 重新赋值头结点t = newNode;}}if (h == null) // 头结点为null// 新生头结点与尾结点h = t = new Node<E>(null);// 赋值头结点head = h;// 赋值尾结点tail = t;}

    public boolean offer(E e) {// 元素不为null
        checkNotNull(e);// 新生一个结点final Node<E> newNode = new Node<E>(e);for (Node<E> t = tail, p = t;;) { // 无限循环// q为p结点的下一个结点Node<E> q = p.next;if (q == null) { // q结点为null// p is last nodeif (p.casNext(null, newNode)) { // 比较并进行替换p结点的next域// Successful CAS is the linearization point// for e to become an element of this queue,// and for newNode to become "live".if (p != t) // 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) // 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赋值为head，否则，赋值为现在的尾结点p = (t != (t = tail)) ? t : head;else// Check for tail updates after two hops.// 重新赋值p结点p = (p != t && t != (t = tail)) ? t : q;}}

    public E poll() {restartFromHead:for (;;) { // 无限循环for (Node<E> h = head, p = h, q;;) { // 保存头结点// item项E item = p.item;if (item != null && p.casItem(item, null)) { // item不为null并且比较并替换item成功// Successful CAS is the linearization point// for item to be removed from this queue.if (p != h) // p不等于h    // hop two nodes at a time// 更新头结点updateHead(h, ((q = p.next) != null) ? q : p); // 返回itemreturn item;}else if ((q = p.next) == null) { // q结点为null// 更新头结点
                    updateHead(h, p);return null;}else if (p == q) // p等于q// 继续循环continue restartFromHead;else// p赋值为qp = q;}}}

    public boolean remove(Object o) {// 元素为null，返回if (o == null) return false;Node<E> pred = null;for (Node<E> p = first(); p != null; p = succ(p)) { // 获取第一个存活的结点// 第一个存活结点的item值E item = p.item;if (item != null &&o.equals(item) &&p.casItem(item, null)) { // 找到item相等的结点，并且将该结点的item设置为null// p的后继结点Node<E> next = succ(p);if (pred != null && next != null) // pred不为null并且next不为null// 比较并替换next域
                    pred.casNext(p, next);return true;}// pred赋值为ppred = p;}return false;}

    Node<E> first() {restartFromHead:for (;;) { // 无限循环，确保成功for (Node<E> h = head, p = h, q;;) {// p结点的item域是否为nullboolean hasItem = (p.item != null);if (hasItem || (q = p.next) == null) { // item不为null或者next域为null// 更新头结点
                    updateHead(h, p);// 返回结点return hasItem ? p : null;}else if (p == q) // p等于q// 继续从头结点开始continue restartFromHead;else// p赋值为qp = q;}}}

    final Node<E> succ(Node<E> p) {// p结点的next域Node<E> next = p.next;// 如果next域为自身，则返回头结点，否则，返回nextreturn (p == next) ? head : next;}

    public int size() {// 计数int count = 0;for (Node<E> p = first(); p != null; p = succ(p)) // 从第一个存活的结点开始往后遍历if (p.item != null) // 结点的item域不为null// Collection.size() spec says to max outif (++count == Integer.MAX_VALUE) // 增加计数，若达到最大值，则跳出循环break;// 返回大小return count;}

package com.hust.grid.leesf.collections;import java.util.concurrent.ConcurrentLinkedQueue;class PutThread extends Thread {private ConcurrentLinkedQueue<Integer> clq;public PutThread(ConcurrentLinkedQueue<Integer> clq) {this.clq = clq;}public void run() {for (int i = 0; i < 10; i++) {try {System.out.println("add " + i);clq.add(i);Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}}
}class GetThread extends Thread {private ConcurrentLinkedQueue<Integer> clq;public GetThread(ConcurrentLinkedQueue<Integer> clq) {this.clq = clq;}public void run() {for (int i = 0; i < 10; i++) {try {System.out.println("poll " + clq.poll());Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}}
}public class ConcurrentLinkedQueueDemo {public static void main(String[] args) {ConcurrentLinkedQueue<Integer> clq = new ConcurrentLinkedQueue<Integer>();PutThread p1 = new PutThread(clq);GetThread g1 = new GetThread(clq);p1.start();g1.start();}
}

add 0
poll null
add 1
poll 0
add 2
poll 1
add 3
poll 2
add 4
poll 3
add 5
poll 4
poll 5
add 6
add 7
poll 6
poll 7
add 8
add 9
poll 8