阻塞隊列—PriorityBlockingQueue源碼分析

作者：一角錢技術 2020-11-24 09:04:55

PriorityBlockingQueue 優先級隊列，線程安全（添加、讀取都進行了加鎖）、無界、讀阻塞的隊列，底層采用的堆結構實現（二叉樹），默認是小根堆，最小的或者最大的元素會一直置頂，每次獲取都取最頂端的數據

前言

PriorityBlockingQueue 優先級隊列，線程安全(添加、讀取都進行了加鎖)、無界、讀阻塞的隊列，底層采用的堆結構實現(二叉樹)，默認是小根堆，最小的或者最大的元素會一直置頂，每次獲取都取最頂端的數據

隊列創建

小根堆

PriorityBlockingQueue<Integer> concurrentLinkedQueue = new PriorityBlockingQueue<Integer>();

大根堆

PriorityBlockingQueue<Integer> concurrentLinkedQueue = new PriorityBlockingQueue<Integer>(10, new Comparator<Integer>() { 
 @Override 
 public int compare(Integer o1, Integer o2) { 
  return o2 - o1; 
 } 
});

應用場景

有任務要執行，可以對任務加一個優先級的權重，這樣隊列會識別出來，對該任務優先進行出隊。

我們來看一個具體例子，例子中定義了一個將要放入“優先阻塞隊列”的任務類，并且定義了一個任務工場類和一個任務執行類，在任務工場類中產生了各種不同優先級的任務，將其添加到隊列中，在任務執行類中，任務被一個個取出并執行。

package com.niuh.queue.priority; 
 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Queue; 
import java.util.Random; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.PriorityBlockingQueue; 
import java.util.concurrent.TimeUnit; 
 
/** 
 * <p> 
 * PriorityBlockingQueue使用示例 
 * </p> 
 */ 
public class PriorityBlockingQueueDemo { 
 
    public static void main(String[] args) throws Exception { 
        Random random = new Random(47); 
        ExecutorService exec = Executors.newCachedThreadPool(); 
        PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>(); 
        exec.execute(new PrioritizedTaskProducer(queue, exec)); // 這里需要注意，往PriorityBlockingQueue中添加任務和取出任務的 
        exec.execute(new PrioritizedTaskConsumer(queue)); // 步驟是同時進行的，因而輸出結果并不一定是有序的 
    } 
} 
 
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> { 
    private Random random = new Random(47); 
    private static int counter = 0; 
    private final int id = counter++; 
    private final int priority; 
 
    protected static List<PrioritizedTask> sequence = new ArrayList<>(); 
 
    public PrioritizedTask(int priority) { 
        this.priority = priority; 
        sequence.add(this); 
    } 
 
    @Override 
    public int compareTo(PrioritizedTask o) { 
        return priority < o.priority ? 1 : (priority > o.priority ? -1 : 0);  // 定義優先級計算方式 
    } 
 
    @Override 
    public void run() { 
        try { 
            TimeUnit.MILLISECONDS.sleep(random.nextInt(250)); 
        } catch (InterruptedException e) { 
        } 
        System.out.println(this); 
    } 
 
    @Override 
    public String toString() { 
        return String.format("[%1$-3d]", priority) + " Task " + id; 
    } 
 
    public String summary() { 
        return "(" + id + ": " + priority + ")"; 
    } 
 
    public static class EndSentinel extends PrioritizedTask { 
        private ExecutorService exec; 
 
        public EndSentinel(ExecutorService exec) { 
            super(-1); 
            this.exec = exec; 
        } 
 
        @Override 
        public void run() { 
            int count = 0; 
            for (PrioritizedTask pt : sequence) { 
                System.out.print(pt.summary()); 
                if (++count % 5 == 0) { 
                    System.out.println(); 
                } 
            } 
            System.out.println(); 
            System.out.println(this + " Calling shutdownNow()"); 
            exec.shutdownNow(); 
        } 
    } 
} 
 
class PrioritizedTaskProducer implements Runnable { 
    private Random random = new Random(47); 
    private Queue<Runnable> queue; 
    private ExecutorService exec; 
 
    public PrioritizedTaskProducer(Queue<Runnable> queue, ExecutorService exec) { 
        this.queue = queue; 
        this.exec = exec; 
    } 
 
    @Override 
    public void run() { 
        for (int i = 0; i < 20; i++) { 
            queue.add(new PrioritizedTask(random.nextInt(10))); // 往PriorityBlockingQueue中添加隨機優先級的任務 
            Thread.yield(); 
        } 
        try { 
            for (int i = 0; i < 10; i++) { 
                TimeUnit.MILLISECONDS.sleep(250); 
                queue.add(new PrioritizedTask(10)); // 往PriorityBlockingQueue中添加優先級為10的任務 
            } 
            for (int i = 0; i < 10; i++) { 
                queue.add(new PrioritizedTask(i));// 往PriorityBlockingQueue中添加優先級為1-10的任務 
            } 
            queue.add(new PrioritizedTask.EndSentinel(exec)); 
        } catch (InterruptedException e) { 
        } 
        System.out.println("Finished PrioritizedTaskProducer"); 
    } 
} 
 
class PrioritizedTaskConsumer implements Runnable { 
    private PriorityBlockingQueue<Runnable> queue; 
 
    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> queue) { 
        this.queue = queue; 
    } 
 
    @Override 
    public void run() { 
        try { 
            while (!Thread.interrupted()) { 
                queue.take().run(); // 任務的消費者，從PriorityBlockingQueue中取出任務執行 
            } 
        } catch (InterruptedException e) { 
        } 
        System.out.println("Finished PrioritizedTaskConsumer"); 
    } 
}

工作原理

PriorityBlockingQueue 是 JDK1.5 的時候出來的一個阻塞隊列。但是該隊列入隊的時候是不會阻塞的，永遠會加到隊尾。下面我們介紹下它的幾個特點：

PriorityBlockingQueue 和 ArrayBlockingQueue 一樣是基于數組實現的，但后者在初始化時需要指定長度，前者默認長度是 11。
該隊列可以說是真正的無界隊列，它在隊列滿的時候會進行擴容，而前面說的無界阻塞隊列其實都有有界，只是界限太大可以忽略(最大值是 2147483647)
該隊列屬于權重隊列，可以理解為它可以進行排序，但是排序不是從小到大排或從大到小排，是基于數組的堆結構(具體如何排下面會進行分析)
出隊方式和前面的也不同，是根據權重來進行出隊，和前面所說隊列中那種先進先出或者先進后出方式不同。
其存入的元素必須實現Comparator，或者在創建隊列的時候自定義Comparator。

注意：

堆結構實際上是一種完全二叉樹。關于二叉樹可以查看《樹、二叉樹、二叉搜索樹的實現和特性》
堆又分為大頂堆和小頂堆。大頂堆中第一個元素肯定是所有元素中最大的，小頂堆中第一個元素是所有元素中最小的。關于二叉堆可以查看《堆和二叉堆的實現和特性》

源碼分析

定義

PriorityBlockingQueue的類繼承關系如下：

其包含的方法定義如下：

成員屬性

從下面的字段我們可以知道，該隊列可以排序，使用顯示鎖來保證操作的原子性，在空隊列時，出隊線程會堵塞等。

/** 
* 默認數組長度 
*/ 
private static final int DEFAULT_INITIAL_CAPACITY = 11; 
 
/** 
 * 最大達容量，分配時超出可能會出現 OutOfMemoryError 異常 
 */ 
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; 
 
/** 
 * 隊列，存儲我們的元素 
 */ 
private transient Object[] queue; 
 
/** 
 * 隊列長度 
 */ 
private transient int size; 
 
/** 
 * 比較器，入隊進行權重的比較 
 */ 
private transient Comparator<? super E> comparator; 
 
/** 
 * 顯示鎖 
 */ 
private final ReentrantLock lock; 
 
/** 
 * 空隊列時進行線程阻塞的 Condition 對象 
 */ 
private final Condition notEmpty;

構造函數

/** 
* 默認構造，使用長度為 11 的數組，比較器為空 
*/ 
public PriorityBlockingQueue() { 
    this(DEFAULT_INITIAL_CAPACITY, null); 
} 
/** 
* 自定義數據長度構造，比較器為空 
*/ 
public PriorityBlockingQueue(int initialCapacity) { 
    this(initialCapacity, null); 
} 
/** 
* 自定義數組長度，可以自定義比較器 
*/ 
public PriorityBlockingQueue(int initialCapacity, 
                             Comparator<? super E> comparator) { 
    if (initialCapacity < 1) 
        throw new IllegalArgumentException(); 
    this.lock = new ReentrantLock(); 
    this.notEmpty = lock.newCondition(); 
    this.comparator = comparator; 
    this.queue = new Object[initialCapacity]; 
} 
/** 
* 構造函數，帶有初始內容的隊列 
*/ 
public PriorityBlockingQueue(Collection<? extends E> c) { 
    this.lock = new ReentrantLock(); 
    this.notEmpty = lock.newCondition(); 
    boolean heapify = true; // true if not known to be in heap order 
    boolean screen = true;  // true if must screen for nulls 
    if (c instanceof SortedSet<?>) { 
        SortedSet<? extends E> ss = (SortedSet<? extends E>) c; 
        this.comparator = (Comparator<? super E>) ss.comparator(); 
        heapify = false; 
    } 
    else if (c instanceof PriorityBlockingQueue<?>) { 
        PriorityBlockingQueue<? extends E> pq = 
            (PriorityBlockingQueue<? extends E>) c; 
        this.comparator = (Comparator<? super E>) pq.comparator(); 
        screen = false; 
        if (pq.getClass() == PriorityBlockingQueue.class) // exact match 
            heapify = false; 
    } 
    Object[] a = c.toArray(); 
    int n = a.length; 
    // If c.toArray incorrectly doesn't return Object[], copy it. 
    if (a.getClass() != Object[].class) 
        a = Arrays.copyOf(a, n, Object[].class); 
    if (screen && (n == 1 || this.comparator != null)) { 
        for (int i = 0; i < n; ++i) 
            if (a[i] == null) 
                throw new NullPointerException(); 
    } 
    this.queue = a; 
    this.size = n; 
    if (heapify) 
        heapify(); 
}

入隊方法

入隊方法，下面可以看到 put 方法最終會調用 offer 方法，所以我們只看 offer 方法即可。

offer(E e)

public void put(E e) { 
    offer(e); // never need to block 
} 
 
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 { 
        //拿到比較器 
        Comparator<? super E> cmp = comparator; 
        //判斷是否有自定義比較器 
        if (cmp == null) 
            //堆上浮 
            siftUpComparable(n, e, array); 
        else 
            //使用自定義比較器進行堆上浮 
            siftUpUsingComparator(n, e, array, cmp); 
        //隊列長度 +1 
        size = n + 1; 
        //喚醒休眠的出隊線程 
        notEmpty.signal(); 
    } finally { 
        //釋放鎖 
        lock.unlock(); 
    } 
    return true; 
}

siftUpComparable(int k, T x, Object[] array)

上浮調整比較器方法的實現

private static <T> void siftUpComparable(int k, T x, Object[] array) { 
        Comparable<? super T> key = (Comparable<? super T>) x; 
        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; 
    }

根據上面的代碼，可以看出這是完全二叉樹在進行上浮調整。調整入隊的元素，找出最小的，將元素排列有序化。簡單理解就是：父節點元素值一定要比它的子節點得小，如果父節點大于子節點了，那就兩者位置進行交換。

入隊圖解

例子：85 添加到二叉堆中(大頂堆)

package com.niuh.queue.priority; 
 
import java.util.Comparator; 
import java.util.concurrent.PriorityBlockingQueue; 
 
/** 
 * <p> 
 * PriorityBlockingQueue 簡單演示 demo 
 * </p> 
 */ 
public class TestPriorityBlockingQueue { 
 
    public static void main(String[] args) throws InterruptedException { 
        // 大頂堆 
        PriorityBlockingQueue<Integer> concurrentLinkedQueue = new PriorityBlockingQueue<Integer>(10, new Comparator<Integer>() { 
            @Override 
            public int compare(Integer o1, Integer o2) { 
                return o2 - o1; 
            } 
        }); 
 
        concurrentLinkedQueue.offer(90); 
        concurrentLinkedQueue.offer(80); 
        concurrentLinkedQueue.offer(70); 
        concurrentLinkedQueue.offer(60); 
        concurrentLinkedQueue.offer(40); 
        concurrentLinkedQueue.offer(30); 
        concurrentLinkedQueue.offer(20); 
        concurrentLinkedQueue.offer(10); 
        concurrentLinkedQueue.offer(50); 
        concurrentLinkedQueue.offer(85); 
        //輸出元素排列 
        concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  ")); 
        //取出元素 
        Integer take = concurrentLinkedQueue.take(); 
        System.out.println(); 
        concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  ")); 
    } 
}

操作的細節分為兩步：

第一步：首先把新元素插入到堆的尾部再說;(新的元素可能是特別大或者特別小，那么要做的一件事情就是重新維護一下堆的所有元素，把新元素挪到這個堆的相應的位置)
第二步：依次向上調整整個堆的結構，就叫 HeapifyUp

85 按照上面講的先插入到堆的尾部，也就是一維數組的尾部，一維數組的尾部的話就上圖的位置，因為這是一個完全二叉樹，所以它的尾部就是50后面這個結點。插進來之后這個時候就破壞了堆，它的每一個結點都要大于它的兒子的這種屬性了，接下來要做的事情就是要把 85 依次地向上浮動，怎么浮動?就是 85 大于它的父親結點，那么就和父親結點進行交換，直到走到根如果大于根的話，就和根也進行交換。

85 再繼續往前走之后，它要和 80 再進行比較，同理可得：也就是說這個結點每次和它的父親比，如果它大于它的父親的話就交換，直到它不再大于它的父親。

出隊方法

入隊列的方法說完后，我們來說說出隊列的方法。PriorityBlockingQueue提供了多種出隊操作的實現來滿足不同情況下的需求，如下：

E take();
E poll();
E poll(long timeout, TimeUnit unit);
E peek()

poll 和 peek 與上面類似，這里不做說明

take()

出隊方法，該方法會阻塞

public E take() throws InterruptedException { 
 //顯示鎖 
    final ReentrantLock lock = this.lock; 
    //可中斷鎖 
    lock.lockInterruptibly(); 
    //結果接收對象 
    E result; 
    try { 
     //判斷隊列是否為空 
        while ( (result = dequeue()) == null) 
         //線程阻塞 
            notEmpty.await(); 
    } finally { 
        lock.unlock(); 
    } 
    return result; 
}

dequeue()

我們再來看看具體出隊方法的實現，dequeue方法

private E dequeue() { 
//長度減少 1 
   int n = size - 1; 
   //判斷隊列中是否有元素 
   if (n < 0) 
       return null; 
   else { 
    //隊列對象 
       Object[] array = queue; 
       //取出第一個元素 
       E result = (E) array[0]; 
       //拿出最后一個元素 
       E x = (E) array[n]; 
       //置空 
       array[n] = null; 
       Comparator<? super E> cmp = comparator; 
       if (cmp == null) 
        //下沉調整 
           siftDownComparable(0, x, array, n); 
       else 
           siftDownUsingComparator(0, x, array, n, cmp); 
       //成功則減少隊列中的元素數量 
       size = n; 
       return result; 
   }

總體就是找到父節點與兩個子節點中最小的一個節點，然后進行交換位置，不斷重復，由上而下的交換。

siftDownComparable(int k, T x, Object[] array, int n)

再來看看下沉比較器方法的實現

private static <T> void siftDownComparable(int k, T x, Object[] array, 
                                               int n) { 
    //判斷隊列長度 
    if (n > 0) { 
        Comparable<? super T> key = (Comparable<? super T>)x; 
        //找到隊列最后一個元素的父節點的索引。 
        int half = n >>> 1;           // loop while a non-leaf 
        while (k < half) { 
         //拿到 k 節點下的左子節點 
            int child = (k << 1) + 1; // assume left child is least 
            //取得子節點對應的值 
            Object c = array[child]; 
            //取得 k 右子節點的索引 
            int right = child + 1; 
            //比較右節點的索引是否小于隊列長度和左右子節點的值進行比較 
            if (right < n && 
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0) 
                c = array[child = right]; 
            //比較父節點值是否大于子節點 
            if (key.compareTo((T) c) <= 0) 
                break; 
            //下面都是元素替換 
            array[k] = c; 
            k = child; 
        } 
        array[k] = key; 
    } 
}

出隊圖解

將堆尾元素替換到頂部(即堆頂被替代刪除掉)

依次從根部向下調整整個堆的結構(一直到堆尾即可) HeapifyDown

例子：90 從二叉堆中刪除(大頂堆)

總結

PriorityBlockingQueue 真的是個神奇的隊列，可以實現優先出隊。最特別的是它只有一個鎖，入隊操作永遠成功，而出隊只有在空隊列的時候才會進行線程阻塞。可以說有一定的應用場景吧，比如：有任務要執行，可以對任務加一個優先級的權重，這樣隊列會識別出來，對該任務優先進行出隊。

責任編輯：姜華來源：今日頭條

PriorityBlockingQueu