【併發隊列】阻塞隊列 LinkedBlockingQueue 源碼解析

1、概述

    LinkedBlockingQueue是基於鏈表的阻塞FIFO隊列，能夠指定一個最大的長度限制以防止過分擴展，未指定狀況下其大小爲Integer.MAX_VALUE。

2、同步策略

    putlock只負責添加，takelock只負責刪除。這樣添加和刪除操做就能夠分開作，這樣也是其吞吐量高於ArrayBlockingQueue的緣由，而對於count則使用一個AtomicInteger來進行同步。可是對於迭代器操做和remove（Object）操做則是不一樣的，須要同時加上兩種鎖才行。對於notFull和notEmpty的條件變量這裏實際也是分開的。

3、可見性

讀者和寫者之間的可見性問題，當元素是隊尾元素時，獲取put鎖，而且count更新，隨後的讀者經過fullLock得到put鎖或得到take鎖來保證其對於隊尾元素的可見性。而後讀取n=count.get()，這個保證得到對第n個元素的可見性。

4、源碼解析

 LinkedBlockingQueue 類中定義的變量有：

/** The capacity bound, or Integer.MAX_VALUE if none */  
private final int capacity;  
  
/** Current number of elements */  
private final AtomicInteger count = new AtomicInteger(0);  
  
/** Head of linked list */  
private transient Node<E> head;  
  
/** Tail of linked list */  
private transient Node<E> last;  
  
/** Lock held by take, poll, etc */  
private final ReentrantLock takeLock = new ReentrantLock();  
  
/** Wait queue for waiting takes */  
private final Condition notEmpty = takeLock.newCondition();  
  
/** Lock held by put, offer, etc */  
private final ReentrantLock putLock = new ReentrantLock();  
  
/** Wait queue for waiting puts */  
private final Condition notFull = putLock.newCondition();

該類中定義了兩個ReentrantLock鎖：putLock和takeLock，分別用於put端和take端。也就是說，生成端和消費端各自獨立擁有一把鎖，避免了讀（take）寫（put）時互相競爭鎖的狀況。

public boolean offer(E e)

原理：在隊尾插入一個元素， 若是隊列沒滿，當即返回true； 若是隊列滿了，當即返回false。

/**
     * 在隊尾插入一個元素， 容量沒滿，能夠當即插入，返回true； 隊列滿了，直接返回false
     * 注：若是使用了限制了容量的隊列，這個方法比add()好，由於add()插入失敗就會拋出異常
     */
    public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final AtomicInteger count = this.count;// 獲取隊列中的元素個數
        if (count.get() == capacity)// 隊列滿了
            return false;
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        putLock.lock();// 獲取入隊鎖
        try {
            if (count.get() < capacity) {// 容量沒滿
                enqueue(e);// 入隊
                c = count.getAndIncrement();// 容量+1，返回舊值（注意）
                if (c + 1 < capacity)// 若是添加元素後的容量，還小於指定容量（說明在插入當前元素後，至少還能夠再插一個元素）
                    notFull.signal();// 喚醒等待notFull條件的其中一個線程
            }
        } finally {
            putLock.unlock();// 釋放入隊鎖
        }
        if (c == 0)// 若是c==0，這是什麼狀況？一開始若是是個空隊列，就會是這樣的值，要注意的是，上邊的c返回的是舊值
            signalNotEmpty();
        return c >= 0;
    }


    /**
     * 建立一個節點，並加入鏈表尾部
     * @param x
     */
    private void enqueue(E x) {
        /*
         * 封裝新節點，並賦給當前的最後一個節點的下一個節點，而後在將這個節點設爲最後一個節點
         */
        last = last.next = new Node<E>(x);
    }

    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();//獲取出隊鎖
        try {
            notEmpty.signal();//喚醒等待notEmpty條件的線程中的一個
        } finally {
            takeLock.unlock();//釋放出隊鎖
        }
    }

public void put(E e)

原理：在隊尾插入一個元素，若是隊列滿了，一直阻塞，直到隊列不滿了或者線程被中斷

/**
     * 在隊尾插一個元素
     * 若是隊列滿了，一直阻塞，直到隊列不滿了或者線程被中斷
     */
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        final ReentrantLock putLock = this.putLock;//入隊鎖
        final AtomicInteger count = this.count;//當前隊列中的元素個數
        putLock.lockInterruptibly();//加鎖
        try {
            while (count.get() == capacity) {//若是隊列滿了 
                /*
                 * 加入notFull等待隊列，直到隊列元素不滿了，
                 * 被其餘線程使用notFull.signal()喚醒
                 */
                notFull.await();
            }
            enqueue(e);//入隊
            c = count.getAndIncrement();//入隊數量+1
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

public E take() 

原理：將隊頭元素出隊，若是隊列空了，一直阻塞，直到隊列不爲空或者線程被中斷

/**
     * 出隊：
     * 若是隊列空了，一直阻塞，直到隊列不爲空或者線程被中斷
     */
    public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;//獲取隊列中的元素總量
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//獲取出隊鎖
        try {
            while (count.get() == 0) {//若是沒有元素，一直阻塞
                /*
                 * 加入等待隊列， 一直等待條件notEmpty（即被其餘線程喚醒）
                 * （喚醒其實就是，有線程將一個元素入隊了，而後調用notEmpty.signal()喚醒其餘等待這個條件的線程，同時隊列也不空了）
                 */
                notEmpty.await();
            }
            x = dequeue();//出隊
            c = count.getAndDecrement();//元素數量-1
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

    /**
     * 從隊列頭部移除一個節點
     */
    private E dequeue() {
        Node<E> h = head;//獲取頭節點：x==null
        Node<E> first = h.next;//將頭節點的下一個節點賦值給first
        h.next = h; // 將當前將要出隊的節點置null（爲了使其作head節點作準備）
        head = first;//將當前將要出隊的節點做爲了頭節點
        E x = first.item;//獲取出隊節點的值
        first.item = null;//將出隊節點的值置空
        return x;
    }

    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

public E poll()

原理：若是沒有元素，直接返回null；若是有元素，出隊

/**
     * 出隊： 
     * 一、若是沒有元素，直接返回null 
     * 二、若是有元素，出隊
     */
    public E poll() {
        final AtomicInteger count = this.count;// 獲取元素數量
        if (count.get() == 0)// 沒有元素
            return null;
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();// 獲取出隊鎖
        try {
            if (count.get() > 0) {// 有元素
                x = dequeue();// 出隊
                // 元素個數-1（注意：該方法是一個無限循環，直到減1成功爲止，且返回舊值）
                c = count.getAndDecrement();
                if (c > 1)// 還有元素（若是舊值c==1的話，那麼經過上邊的操做以後，隊列就空了）
                    notEmpty.signal();// 喚醒等待在notEmpty隊列中的其中一條線程
            }
        } finally {
            takeLock.unlock();// 釋放出隊鎖
        }
        if (c == capacity)// c == capacity是怎麼發生的？若是隊列是一個滿隊列，注意：上邊的c返回的是舊值
            signalNotFull();
        return x;
    }

5、總結

一、具體入隊與出隊的原理圖：

圖中每個節點前半部分表示封裝的數據x，後邊的表示指向的下一個引用。

1.一、初始化

 初始化以後，初始化一個數據爲null，且head和last節點都是這個節點。

1.二、入隊兩個元素事後

1.三、出隊一個元素後

表面上看，只是將頭節點的next指針指向了要刪除的x1.next，事實上這樣我覺的就徹底能夠，可是jdk其實是將原來的head節點刪除了，而上邊看到的這個head節點，正是剛剛出隊的x1節點，只是其值被置空了。

二、三種入隊對比：

• offer(E e)：若是隊列沒滿，當即返回true； 若是隊列滿了，當即返回false-->不阻塞
• put(E e)：若是隊列滿了，一直阻塞，直到隊列不滿了或者線程被中斷-->阻塞
• offer(E e, long timeout, TimeUnit unit)：在隊尾插入一個元素,，若是隊列已滿，則進入等待，直到出現如下三種狀況：-->阻塞
  • 被喚醒
  • 等待時間超時
  • 當前線程被中斷

三、三種出隊對比：

• poll()：若是沒有元素，直接返回null；若是有元素，出隊
• take()：若是隊列空了，一直阻塞，直到隊列不爲空或者線程被中斷-->阻塞
• poll(long timeout, TimeUnit unit)：若是隊列不空，出隊；若是隊列已空且已經超時，返回null；若是隊列已空且時間未超時，則進入等待，直到出現如下三種狀況：
  • 被喚醒
  • 等待時間超時
  • 當前線程被中斷

四、ArrayBlockingQueue與LinkedBlockingQueue對比

• ArrayBlockingQueue：
  • 一個對象數組+一把鎖+兩個條件
  • 入隊與出隊都用同一把鎖
  • 在只有入隊高併發或出隊高併發的狀況下，由於操做數組，且不須要擴容，性能很高
  • 採用了數組，必須指定大小，即容量有限
• LinkedBlockingQueue：
  • 一個單向鏈表+兩把鎖+兩個條件
  • 兩把鎖，一把用於入隊，一把用於出隊，有效的避免了入隊與出隊時使用一把鎖帶來的競爭。
  • 在入隊與出隊都高併發的狀況下，性能比ArrayBlockingQueue高不少
  • 採用了鏈表，最大容量爲整數最大值，可看作容量無限