點選上方“芋道原始碼”，選擇“置頂公眾號”

技術文章第一時間送達！

摘要: 原創出處 https://www.cnblogs.com/mantu/p/5802393.html 「mantu」歡迎轉載，保留摘要，謝謝！

寫這篇文章源於我經歷過的一次生產事故，在某家公司的時候，有個服務會收集業務系統的日誌，此服務的開發人員在給業務系統的sdk中就因為使用了LinkedList，又沒有做併發控制，就造成了此服務經常不能正常收集到業務系統的日誌(丟日誌以及日誌上報的執行緒停止執行)。看一下add()方法的原始碼，我們就可以知道原因了：

demo 　Lesson2LinkedListThreads 展示了在多執行緒且沒有做併發控制的環境下，size的值遠遠大於了佇列的實際值，100個執行緒，每個新增1000個元素，最後實際只加進去2030個元素：

List的變數size值為：88371
第2031個元素取出為null

解決方案，使用鎖或者使用ConcurrentLinkedQueue、LinkedBlockingQueue等支援新增元素為原子操作的佇列。

上一節我們已經分析過LinkedBlockingQueue的put等方法的原始碼，是使用ReentrantLock來實現的新增元素原子操作。我們再簡單看一下高併發queue的add和offer(）方法，方法中使用了CAS來實現的無鎖的原子操作：

public boolean add(E e) {
　　　　   return offer(e);
　　   }

public boolean offer(E e) {
        checkNotNull(e);
        final Node newNode = new Node(e);

        for (Node t = tail, p = t;;) {
            Node 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;
        }
    }

　　接下來，我們再利用高併發queue對上面的demo進行改造，大家只要改變demo中的內容，講下麵兩行的註釋內容顛倒，即可發現沒有丟失任何的元素：

public static LinkedList list = new LinkedList();
//public static ConcurrentLinkedQueue list = new ConcurrentLinkedQueue();

再看一下高效能queue的poll()方法，才覺得NB，取元素的方法也用CAS實現了原子操作，因此在實際使用的過程中，當我們在不那麼在意元素處理順序的情況下，佇列元素的消費者，完全可以是多個，不會丟任何資料：

    public E poll() {
        restartFromHead:
        for (;;) {
            for (Node h = head, p = h, q;;) {
                E item = p.item;

                if (item != null && p.casItem(item, null)) {
                    // Successful CAS is the linearization point
                    // for item to be removed from this queue.
                    if (p != h) // hop two nodes at a time
                        updateHead(h, ((q = p.next) != null) ? q : p);
                    return item;
                }
                else if ((q = p.next) == null) {
                    updateHead(h, p);
                    return null;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }

關於ConcurrentLinkedQueue和LinkedBlockingQueue：

1.LinkedBlockingQueue是使用鎖機制，ConcurrentLinkedQueue是使用CAS演演算法，雖然LinkedBlockingQueue的底層獲取鎖也是使用的CAS演演算法

2.關於取元素，ConcurrentLinkedQueue不支援阻塞去取元素，LinkedBlockingQueue支援阻塞的take()方法，如若大家需要ConcurrentLinkedQueue的消費者產生阻塞效果，需要自行實現

3.關於插入元素的效能，從字面上和程式碼簡單的分析來看ConcurrentLinkedQueue肯定是最快的，但是這個也要看具體的測試場景，我做了兩個簡單的demo做測試，測試的結果如下，兩個的效能差不多，但在實際的使用過程中，尤其在多cpu的伺服器上，有鎖和無鎖的差距便體現出來了，ConcurrentLinkedQueue會比LinkedBlockingQueue快很多：

demo Lesson2ConcurrentLinkedQueuePerform:在使用ConcurrentLinkedQueue的情況下100個執行緒迴圈增加的元素數為：33828193

demo Lesson2LinkedBlockingQueuePerform:在使用LinkedBlockingQueue的情況下100個執行緒迴圈增加的元素數為：33827382

666. 彩蛋