一、           基本结构

在并发编程中，有时需要使用线程安全的队列。实现一个线程安全的队列有两种方式：一是使用阻塞方式，在入队、出队、读取加锁；另一种可以使用非阻塞方式，即利用CAS。Doug Lee就使用了非阻塞方式设计了ConcurrentLinkedList，类图如下：

http://s15/mw690/002i2qf8zy7jTyRs7Cude&690

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

下面为了简化，我们自己写了一个类似的队列。

private static class MyConcurrentLinkedQueue {
private transient volatile MyNodehead;
    private transient volatile MyNodetail;

    private static long headOffset;
    private static long tailOffset;

    private static Unsafe unsafe;
    static {
try {
            Field field =Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
unsafe = (Unsafe) field.get(null);
Class<?> k = MyConcurrentLinkedQueue.class;
headOffset= unsafe.objectFieldOffset
                    (k.getDeclaredField("head"));
tailOffset= unsafe.objectFieldOffset
                    (k.getDeclaredField("tail"));
} catch (NoSuchFieldException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
}
    }

MyConcurrentLinkedQueue() {
head = tail = new MyNode(null);
}

public void setFirstTail(MyNode node) {
tail.casNext(null, node);
System.out.println();
}

final void updateHead(MyNode h, MyNode p) {
if (h != p &&casHead(h, p))
h.lazySetNext(h);
}

public booleancasHead(MyNodecmp, MyNodeval) {
return unsafe.compareAndSwapObject(this, headOffset, cmp, val);
}

public booleancasTail(MyNodecmp, MyNodeval) {
return unsafe.compareAndSwapObject(this, tailOffset, cmp, val);
}

public booleanoffer(String e) {
final MyNodenewNode = new MyNode(e);
        for (MyNode t = tail, p = t;;) {
MyNode q = p.next;
            if (q == null) {
// p is last node
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".
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
}
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;
}
    }

public String peek() {
restartFromHead:
for (;;) {
for(MyNode h = head, p = h, q;;) {
                String item = p.item;
                if (item != null || (q = p.next) == null) {
updateHead(h, p);
                    return item;
}
else if (p == q)
continue restartFromHead;
                else
p = q;
}
        }
    }
}

private static class MyNode {
private volatile String item;
    private volatile MyNodenext;
    private static Unsafe unsafe;

    private static long itemOffset;
    private static long nextOffset;

MyNode(String item) {
unsafe.putObject(this, itemOffset, item);
}

public booleancasNext(MyNodecmp, MyNodeval) {
return unsafe.compareAndSwapObject(this, nextOffset, cmp, val);
}

void lazySetNext(MyNodeval) {
unsafe.putOrderedObject(this, nextOffset, val);
}

public void setNext(MyNode next) {
this.next= next;
}

public void setItem(String s) {
item = s;
}
public String getItem() {
return item;
}

static {
try {
            Field field =Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
unsafe = (Unsafe) field.get(null);
Class<?> k = MyNode.class;

itemOffset= unsafe.objectFieldOffset(k.getDeclaredField("item"));
nextOffset= unsafe.objectFieldOffset(k.getDeclaredField("next"));
} catch (NoSuchFieldException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
}
    }
}

二、           入队

http://s7/mw690/002i2qf8zy7jTySnnYa96&690

         由上面代码可以发现，offer中不是每次都设置新插入节点为tail。而是当前tail和新插入节点中间隔了一个节点才会重新调整tail所指引用。这样的好处是减少了每次cas设置tail的次数。

         测试代码：

@Test
public void test() {

ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
queue.add("1");
System.out.println(queue.peek());
queue.add("2");
}
@Test
public void testSimulatedMyLinkedQueue() throws NoSuchFieldException, IllegalAccessException {

MyConcurrentLinkedQueuemyLinkedQueue = new MyConcurrentLinkedQueue();
myLinkedQueue.offer("1");
System.out.println(myLinkedQueue.peek());
myLinkedQueue.offer("2");
System.out.println();
}

         这里有一个奇怪的现象无法解释：在test中ConcurrentLinkedQueue执行第一个add，从而执行p.casNext方法，导致head和tail居然分道扬镳，cas没有成功对tail的next赋值为新的newNode，反而使得head的地址成为了新的newNode，tail的next居然是原tail本身。

所以我们在testSimulatedMyLinkedQueue模拟了这一过程，却无法还原这一奇怪现象。p.casNext成功将head和tail所在地址的next改为newNode，item为null。后面的peek方法正好又重新整合了队列，使得head的item为"1"，next仍然为空，原head所在地址（也就是tail）的值改为item为null，next为该地址本身；而下一个offer("2")中，使得奇怪的tail由item为null，next为本身变成新增节点，从此开始队列一切正常。

三、           出队

http://s4/mw690/002i2qf8zy7jTyT1gpdc3&690

从图中可知，并不是每次出队时都更新head节点，当head节点里有元素时，直接弹出head 节点里的元素，而不会更新head节点。只有当head节点里没有元素时，出队操作才会更新head 节点。这种做法也是通过hops变量来减少使用CAS更新head节点的消耗，从而提高出队效率。