private transient volatile Node<E> head;private transient volatile Node<E> tail;// 默认的无常构造使头节点和尾节点都指向一个值为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 = newNode;}}if (h == null)h = t = new Node<E>(null);head = h;tail = t;}// 内部类Node，使用Unsafe类来保证CAS操作的原子性private static class Node<E> {// 元素值volatile E item;// 下一个节点volatile Node<E> next;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 mechanicsprivate static final sun.misc.Unsafe UNSAFE;private static final long itemOffset;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 boolean offer(E e) {// 校验提供的元素e，如果为null时则会抛出异常checkNotNull(e);// 使用原素e来构建一个新的节点final Node<E> newNode = new Node<E>(e);for (Node<E> t = tail, p = t;;) {Node<E> q = p.next;if (q == null) {// p is last node q为null则表明p是最后一个节点  ，下一步使用cas操作，将新建的节点赋给pif (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)// 由于多线程操作，将head节点的next设置为自己，因此会出现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.// 如果之前的尾节点不是当前的尾节点（最新的尾节点），则将t重设为最新的尾节点（因为tail是内存可见的，其他线程操作后，当前线程是可以看得见的），p = (t != (t = tail)) ? t : head;else// Check for tail updates after two hops.  循环找尾节点p = (p != t && t != (t = tail)) ? t : q;}}// add方法内部也是走的offer方法public boolean add(E e) {return offer(e);}

    public E poll() { // goto标记  和循环充的continue restartFromHead关联，即重新走循环语句restartFromHead:for (;;) {for (Node<E> h = head, p = h, q;;) {// 获取当前节点E item = p.item;// CAS操作将当前head节点设置为nullif (item != null && p.casItem(item, null)) {// Successful CAS is the linearization point// for item to be removed from this queue.// 设置成功后（判断是否成功，根据p和h，未修改之前时，p==h，修改成功后，p！=h），成功后，充值head节点if (p != h) // hop two nodes at a timeupdateHead(h, ((q = p.next) != null) ? q : p);return item;}// p.next == null 表明链表为空else if ((q = p.next) == null) {updateHead(h, p);return null;}else if (p == q)continue restartFromHead;elsep = q;}}}    

    transient Node<E> head;private transient Node<E> last;/** 出队锁*/private final ReentrantLock takeLock = new ReentrantLock();/** 非空条件变量 */private final Condition notEmpty = takeLock.newCondition();/** 入队锁*/private final ReentrantLock putLock = new ReentrantLock();/** 非满条件变量*/private final Condition notFull = putLock.newCondition();

    public boolean offer(E e) {// 如果参数e为null时，则会抛出异常if (e == null) throw new NullPointerException();// 获取容量count，AtomicInteger类型对象final AtomicInteger count = this.count;// 判断当前容量是否已满，满的话返回false，不能入队if (count.get() == capacity)return false;int c = -1;// 将参数e构建为节点对象Node<E> node = new Node<E>(e);// 获取入队锁final ReentrantLock putLock = this.putLock;putLock.lock();try {if (count.get() < capacity) {//入队enqueue(node);// 节点数量++c = count.getAndIncrement();// 释放非满信号，可以继续入队if (c + 1 < capacity)notFull.signal();}} finally {putLock.unlock();}if (c == 0)signalNotEmpty();return c >= 0;}// 如对，链表的最后一个节点private void enqueue(Node<E> node) {// assert putLock.isHeldByCurrentThread();// assert last.next == null;last = last.next = node;} 

    public void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();// Note: convention in all put/take/etc is to preset local var// holding count negative to indicate failure unless set.int c = -1;Node<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;putLock.lockInterruptibly();try {// 如果已满，则阻塞线程，等待相应唤醒，唤醒之后会继续判断是否已满（防止伪共享出现）while (count.get() == capacity) {notFull.await();}enqueue(node);c = count.getAndIncrement();if (c + 1 < capacity)// 发出唤醒其他线程可以入队的信号notFull.signal();} finally {putLock.unlock();}if (c == 0)signalNotEmpty();}

    /** 队列数组 */final Object[] items;/** 出队索引 */int takeIndex;/** 入队索引 */int putIndex;/** 队列元素个数 */int count;/** Concurrency control uses the classic two-condition algorithm* found in any textbook.*//** 一个独占锁 */final ReentrantLock lock;/** 条件变量 */private final Condition notEmpty;/** 条件变量 */private final Condition notFull;

    public boolean offer(E e) {// 校验元素非空checkNotNull(e);final ReentrantLock lock = this.lock;// 加锁lock.lock();try {// 已满则返回falseif (count == items.length)return false;else {enqueue(e);return true;}} finally {lock.unlock();}}

    private void enqueue(E x) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;// 向入队索引处插入新元素final Object[] items = this.items;items[putIndex] = x;// 插入之后，入队索引自增if (++putIndex == items.length)putIndex = 0;count++;notEmpty.signal();}

    public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return (count == 0) ? null : dequeue();} finally {lock.unlock();}}public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)// 阻塞当前线程notEmpty.await();return dequeue();} finally {lock.unlock();}}

    // add方法内部也是调用的offer方法 public boolean add(E e) {return offer(e);}public boolean offer(E e) {if (e == null)throw new NullPointerException();// 获取独占锁final ReentrantLock lock = this.lock;lock.lock();int n, cap;Object[] array;// 判断您是否需要进行扩容while ((n = size) >= (cap = (array = queue).length))tryGrow(array, cap);try {// 获取当前的比较器，默认比较器为null（如果构建队列的时候没有提供比较器，则为null）Comparator<? super E> cmp = comparator;if (cmp == null)siftUpComparable(n, e, array);elsesiftUpUsingComparator(n, e, array, cmp);size = n + 1;notEmpty.signal();} finally {lock.unlock();}return true;}// 扩容private void tryGrow(Object[] array, int oldCap) {// 在扩容开始，先释放锁，是为了性能考虑，扩容是需要时间的，如果在扩容的同事占用锁，会降低并发性，所以为了提供并发性，使用CAS操作来保证只有一个线程可以进行扩容，让其他线程可以入队和出队lock.unlock(); // must release and then re-acquire main lock// 扩容之后的数组 ，具体对象等计算出新的大小后会赋值Object[] newArray = null;// allocationSpinLock为0，表示没有线程进行扩容，使用CAS操作设置该变量为1，则表示有线程正在进行扩容， 也就是锁CAS操作成功则进行扩容if (allocationSpinLock == 0 &&UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,0, 1)) {try {// 当容量小的时候，扩容增速块，大64后，扩容为50%int newCap = oldCap + ((oldCap < 64) ?(oldCap + 2) : // grow faster if small(oldCap >> 1));// 如果扩容后，大于Maxinteger-8，则设置默认最大容量if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflowint minCap = oldCap + 1;if (minCap < 0 || minCap > MAX_ARRAY_SIZE)throw new OutOfMemoryError();newCap = MAX_ARRAY_SIZE;}// 新建指定大小的列表数组if (newCap > oldCap && queue == array)newArray = new Object[newCap];} finally {// 同时将allocationSpinLock重新设置为0，表名没有正在进行扩容的线程allocationSpinLock = 0;}}// 这一个判断是当第一个线程CAS成功之后，第二个线程也进入扩容节点，则让第二线程让出cpu，让第一线程尽快执行完扩容if (newArray == null) // back off if another thread is allocatingThread.yield();lock.lock();// 扩容成功之后，将旧数组中的数据复制到新数组中if (newArray != null && queue == array) {queue = newArray;System.arraycopy(array, 0, newArray, 0, oldCap);}}// 默认比较器的入队操作，也就是建堆算法private static <T> void siftUpComparable(int k, T x, Object[] array) {// 新增的元素都是Compareable的子类Comparable<? super T> key = (Comparable<? super T>) x;// k为之前队列个数，如果原来队列元素大于0，则需要判断当前新增元素的位置，否则，直接入队while (k > 0) {int parent = (k - 1) >>> 1;Object e = array[parent];if (key.compareTo((T) e) >= 0)break;array[k] = e;k = parent;}array[k] = key;}private static <T> void siftUpUsingComparator(int k, T x, Object[] array,Comparator<? super T> cmp) {while (k > 0) {int parent = (k - 1) >>> 1;Object e = array[parent];if (cmp.compare(x, (T) e) >= 0)break;array[k] = e;k = parent;}array[k] = x;}

 1    // 默认比较器的情况 2    private static <T> void siftUpComparable(int k, T x, Object[] array) {3         Comparable<? super T> key = (Comparable<? super T>) x;4         while (k > 0) {5             int parent = (k - 1) >>> 1;6             Object e = array[parent];7             if (key.compareTo((T) e) >= 0)8                 break;9             array[k] = e;
10             k = parent;
11         }
12         array[k] = key;
13     }