av激情亚洲男人的天堂国语,日韩欧美精品一中文字幕,无码av一区二区三区无码,国产又色又爽又刺激的a片,国产又色又爽又刺激的a片

阻塞隊(duì)列—PriorityBlockingQueue源碼分析

前言

PriorityBlockingQueue 優(yōu)先級隊(duì)列，線程安全(添加、讀取都進(jìn)行了加鎖)、無界、讀阻塞的隊(duì)列，底層采用的堆結(jié)構(gòu)實(shí)現(xiàn)(二叉樹)，默認(rèn)是小根堆，最小的或者最大的元素會一直置頂，每次獲取都取最頂端的數(shù)據(jù)

隊(duì)列創(chuàng)建

小根堆

 
 
 
 
      
  
  
  
  1. PriorityBlockingQueue concurrentLinkedQueue = new PriorityBlockingQueue(); 
    2.

大根堆

 
 
 
 
      
  
  
  
  1. PriorityBlockingQueue concurrentLinkedQueue = new PriorityBlockingQueue(10, new Comparator() { 
    2.   
  
  
  
  2.  @Override 
    3.   
  
  
  
  3.  public int compare(Integer o1, Integer o2) { 
    4.   
  
  
  
  4.   return o2 - o1; 
    5.   
  
  
  
  5.  } 
    6.   
  
  
  
  6. }); 
    7.

 應(yīng)用場景

有任務(wù)要執(zhí)行，可以對任務(wù)加一個(gè)優(yōu)先級的權(quán)重，這樣隊(duì)列會識別出來，對該任務(wù)優(yōu)先進(jìn)行出隊(duì)。

我們來看一個(gè)具體例子，例子中定義了一個(gè)將要放入“優(yōu)先阻塞隊(duì)列”的任務(wù)類，并且定義了一個(gè)任務(wù)工場類和一個(gè)任務(wù)執(zhí)行類，在任務(wù)工場類中產(chǎn)生了各種不同優(yōu)先級的任務(wù)，將其添加到隊(duì)列中，在任務(wù)執(zhí)行類中，任務(wù)被一個(gè)個(gè)取出并執(zhí)行。

 
 
 
 
      
  
  
  
  1. package com.niuh.queue.priority; 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import java.util.ArrayList; 
    4.   
  
  
  
  4. import java.util.List; 
    5.   
  
  
  
  5. import java.util.Queue; 
    6.   
  
  
  
  6. import java.util.Random; 
    7.   
  
  
  
  7. import java.util.concurrent.ExecutorService; 
    8.   
  
  
  
  8. import java.util.concurrent.Executors; 
    9.   
  
  
  
  9. import java.util.concurrent.PriorityBlockingQueue; 
    10.   
  
  
  
  10. import java.util.concurrent.TimeUnit; 
    11.   
  
  
  
  11.  
    12.   
  
  
  
  12. /** 
    13.   
  
  
  
  13.  * 
     
    14.   
  
  
  
  14.  * PriorityBlockingQueue使用示例 
    15.   
  
  
  
  15.  * 
     
    16.   
  
  
  
  16.  */ 
    17.   
  
  
  
  17. public class PriorityBlockingQueueDemo { 
    18.   
  
  
  
  18.  
    19.   
  
  
  
  19.     public static void main(String[] args) throws Exception { 
    20.   
  
  
  
  20.         Random random = new Random(47); 
    21.   
  
  
  
  21.         ExecutorService exec = Executors.newCachedThreadPool(); 
    22.   
  
  
  
  22.         PriorityBlockingQueue queue = new PriorityBlockingQueue<>(); 
    23.   
  
  
  
  23.         exec.execute(new PrioritizedTaskProducer(queue, exec)); // 這里需要注意，往PriorityBlockingQueue中添加任務(wù)和取出任務(wù)的 
    24.   
  
  
  
  24.         exec.execute(new PrioritizedTaskConsumer(queue)); // 步驟是同時(shí)進(jìn)行的，因而輸出結(jié)果并不一定是有序的 
    25.   
  
  
  
  25.     } 
    26.   
  
  
  
  26. } 
    27.   
  
  
  
  27.  
    28.   
  
  
  
  28. class PrioritizedTask implements Runnable, Comparable { 
    29.   
  
  
  
  29.     private Random random = new Random(47); 
    30.   
  
  
  
  30.     private static int counter = 0; 
    31.   
  
  
  
  31.     private final int id = counter++; 
    32.   
  
  
  
  32.     private final int priority; 
    33.   
  
  
  
  33.  
    34.   
  
  
  
  34.     protected static List sequence = new ArrayList<>(); 
    35.   
  
  
  
  35.  
    36.   
  
  
  
  36.     public PrioritizedTask(int priority) { 
    37.   
  
  
  
  37.         this.priority = priority; 
    38.   
  
  
  
  38.         sequence.add(this); 
    39.   
  
  
  
  39.     } 
    40.   
  
  
  
  40.  
    41.   
  
  
  
  41.     @Override 
    42.   
  
  
  
  42.     public int compareTo(PrioritizedTask o) { 
    43.   
  
  
  
  43.         return priority < o.priority ? 1 : (priority > o.priority ? -1 : 0);  // 定義優(yōu)先級計(jì)算方式 
    44.   
  
  
  
  44.     } 
    45.   
  
  
  
  45.  
    46.   
  
  
  
  46.     @Override 
    47.   
  
  
  
  47.     public void run() { 
    48.   
  
  
  
  48.         try { 
    49.   
  
  
  
  49.             TimeUnit.MILLISECONDS.sleep(random.nextInt(250)); 
    50.   
  
  
  
  50.         } catch (InterruptedException e) { 
    51.   
  
  
  
  51.         } 
    52.   
  
  
  
  52.         System.out.println(this); 
    53.   
  
  
  
  53.     } 
    54.   
  
  
  
  54.  
    55.   
  
  
  
  55.     @Override 
    56.   
  
  
  
  56.     public String toString() { 
    57.   
  
  
  
  57.         return String.format("[%1$-3d]", priority) + " Task " + id; 
    58.   
  
  
  
  58.     } 
    59.   
  
  
  
  59.  
    60.   
  
  
  
  60.     public String summary() { 
    61.   
  
  
  
  61.         return "(" + id + ": " + priority + ")"; 
    62.   
  
  
  
  62.     } 
    63.   
  
  
  
  63.  
    64.   
  
  
  
  64.     public static class EndSentinel extends PrioritizedTask { 
    65.   
  
  
  
  65.         private ExecutorService exec; 
    66.   
  
  
  
  66.  
    67.   
  
  
  
  67.         public EndSentinel(ExecutorService exec) { 
    68.   
  
  
  
  68.             super(-1); 
    69.   
  
  
  
  69.             this.exec = exec; 
    70.   
  
  
  
  70.         } 
    71.   
  
  
  
  71.  
    72.   
  
  
  
  72.         @Override 
    73.   
  
  
  
  73.         public void run() { 
    74.   
  
  
  
  74.             int count = 0; 
    75.   
  
  
  
  75.             for (PrioritizedTask pt : sequence) { 
    76.   
  
  
  
  76.                 System.out.print(pt.summary()); 
    77.   
  
  
  
  77.                 if (++count % 5 == 0) { 
    78.   
  
  
  
  78.                     System.out.println(); 
    79.   
  
  
  
  79.                 } 
    80.   
  
  
  
  80.             } 
    81.   
  
  
  
  81.             System.out.println(); 
    82.   
  
  
  
  82.             System.out.println(this + " Calling shutdownNow()"); 
    83.   
  
  
  
  83.             exec.shutdownNow(); 
    84.   
  
  
  
  84.         } 
    85.   
  
  
  
  85.     } 
    86.   
  
  
  
  86. } 
    87.   
  
  
  
  87.  
    88.   
  
  
  
  88. class PrioritizedTaskProducer implements Runnable { 
    89.   
  
  
  
  89.     private Random random = new Random(47); 
    90.   
  
  
  
  90.     private Queue queue; 
    91.   
  
  
  
  91.     private ExecutorService exec; 
    92.   
  
  
  
  92.  
    93.   
  
  
  
  93.     public PrioritizedTaskProducer(Queue queue, ExecutorService exec) { 
    94.   
  
  
  
  94.         this.queue = queue; 
    95.   
  
  
  
  95.         this.exec = exec; 
    96.   
  
  
  
  96.     } 
    97.   
  
  
  
  97.  
    98.   
  
  
  
  98.     @Override 
    99.   
  
  
  
  99.     public void run() { 
    100.   
  
  
  
  100.         for (int i = 0; i < 20; i++) { 
    101.   
  
  
  
  101.             queue.add(new PrioritizedTask(random.nextInt(10))); // 往PriorityBlockingQueue中添加隨機(jī)優(yōu)先級的任務(wù) 
    102.   
  
  
  
  102.             Thread.yield(); 
    103.   
  
  
  
  103.         } 
    104.   
  
  
  
  104.         try { 
    105.   
  
  
  
  105.             for (int i = 0; i < 10; i++) { 
    106.   
  
  
  
  106.                 TimeUnit.MILLISECONDS.sleep(250); 
    107.   
  
  
  
  107.                 queue.add(new PrioritizedTask(10)); // 往PriorityBlockingQueue中添加優(yōu)先級為10的任務(wù) 
    108.   
  
  
  
  108.             } 
    109.   
  
  
  
  109.             for (int i = 0; i < 10; i++) { 
    110.   
  
  
  
  110.                 queue.add(new PrioritizedTask(i));// 往PriorityBlockingQueue中添加優(yōu)先級為1-10的任務(wù) 
    111.   
  
  
  
  111.             } 
    112.   
  
  
  
  112.             queue.add(new PrioritizedTask.EndSentinel(exec)); 
    113.   
  
  
  
  113.         } catch (InterruptedException e) { 
    114.   
  
  
  
  114.         } 
    115.   
  
  
  
  115.         System.out.println("Finished PrioritizedTaskProducer"); 
    116.   
  
  
  
  116.     } 
    117.   
  
  
  
  117. } 
    118.   
  
  
  
  118.  
    119.   
  
  
  
  119. class PrioritizedTaskConsumer implements Runnable { 
    120.   
  
  
  
  120.     private PriorityBlockingQueue queue; 
    121.   
  
  
  
  121.  
    122.   
  
  
  
  122.     public PrioritizedTaskConsumer(PriorityBlockingQueue queue) { 
    123.   
  
  
  
  123.         this.queue = queue; 
    124.   
  
  
  
  124.     } 
    125.   
  
  
  
  125.  
    126.   
  
  
  
  126.     @Override 
    127.   
  
  
  
  127.     public void run() { 
    128.   
  
  
  
  128.         try { 
    129.   
  
  
  
  129.             while (!Thread.interrupted()) { 
    130.   
  
  
  
  130.                 queue.take().run(); // 任務(wù)的消費(fèi)者，從PriorityBlockingQueue中取出任務(wù)執(zhí)行 
    131.   
  
  
  
  131.             } 
    132.   
  
  
  
  132.         } catch (InterruptedException e) { 
    133.   
  
  
  
  133.         } 
    134.   
  
  
  
  134.         System.out.println("Finished PrioritizedTaskConsumer"); 
    135.   
  
  
  
  135.     } 
    136.   
  
  
  
  136. } 
    137.

 工作原理

PriorityBlockingQueue 是 JDK1.5 的時(shí)候出來的一個(gè)阻塞隊(duì)列。但是該隊(duì)列入隊(duì)的時(shí)候是不會阻塞的，永遠(yuǎn)會加到隊(duì)尾。下面我們介紹下它的幾個(gè)特點(diǎn)：

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

注意：

  1. 堆結(jié)構(gòu)實(shí)際上是一種完全二叉樹。關(guān)于二叉樹可以查看 《樹、二叉樹、二叉搜索樹的實(shí)現(xiàn)和特性》
  2. 堆又分為大頂堆和小頂堆 。大頂堆中第一個(gè)元素肯定是所有元素中最大的，小頂堆中第一個(gè)元素是所有元素中最小的。關(guān)于二叉堆可以查看《堆和二叉堆的實(shí)現(xiàn)和特性》

源碼分析

定義

PriorityBlockingQueue的類繼承關(guān)系如下：

其包含的方法定義如下：

成員屬性

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

 
 
 
 
      
  
  
  
  1. /** 
    2.   
  
  
  
  2. * 默認(rèn)數(shù)組長度 
    3.   
  
  
  
  3. */ 
    4.   
  
  
  
  4. private static final int DEFAULT_INITIAL_CAPACITY = 11; 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6. /** 
    7.   
  
  
  
  7.  * 最大達(dá)容量，分配時(shí)超出可能會出現(xiàn) OutOfMemoryError 異常 
    8.   
  
  
  
  8.  */ 
    9.   
  
  
  
  9. private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; 
    10.   
  
  
  
  10.  
    11.   
  
  
  
  11. /** 
    12.   
  
  
  
  12.  * 隊(duì)列，存儲我們的元素 
    13.   
  
  
  
  13.  */ 
    14.   
  
  
  
  14. private transient Object[] queue; 
    15.   
  
  
  
  15.  
    16.   
  
  
  
  16. /** 
    17.   
  
  
  
  17.  * 隊(duì)列長度 
    18.   
  
  
  
  18.  */ 
    19.   
  
  
  
  19. private transient int size; 
    20.   
  
  
  
  20.  
    21.   
  
  
  
  21. /** 
    22.   
  
  
  
  22.  * 比較器，入隊(duì)進(jìn)行權(quán)重的比較 
    23.   
  
  
  
  23.  */ 
    24.   
  
  
  
  24. private transient Comparator comparator; 
    25.   
  
  
  
  25.  
    26.   
  
  
  
  26. /** 
    27.   
  
  
  
  27.  * 顯示鎖 
    28.   
  
  
  
  28.  */ 
    29.   
  
  
  
  29. private final ReentrantLock lock; 
    30.   
  
  
  
  30.  
    31.   
  
  
  
  31. /** 
    32.   
  
  
  
  32.  * 空隊(duì)列時(shí)進(jìn)行線程阻塞的 Condition 對象 
    33.   
  
  
  
  33.  */ 
    34.   
  
  
  
  34. private final Condition notEmpty; 
    35.

 構(gòu)造函數(shù)

 
 
 
 
      
  
  
  
  1. /** 
    2.   
  
  
  
  2. * 默認(rèn)構(gòu)造，使用長度為 11 的數(shù)組，比較器為空 
    3.   
  
  
  
  3. */ 
    4.   
  
  
  
  4. public PriorityBlockingQueue() { 
    5.   
  
  
  
  5.     this(DEFAULT_INITIAL_CAPACITY, null); 
    6.   
  
  
  
  6. } 
    7.   
  
  
  
  7. /** 
    8.   
  
  
  
  8. * 自定義數(shù)據(jù)長度構(gòu)造，比較器為空 
    9.   
  
  
  
  9. */ 
    10.   
  
  
  
  10. public PriorityBlockingQueue(int initialCapacity) { 
    11.   
  
  
  
  11.     this(initialCapacity, null); 
    12.   
  
  
  
  12. } 
    13.   
  
  
  
  13. /** 
    14.   
  
  
  
  14. * 自定義數(shù)組長度，可以自定義比較器 
    15.   
  
  
  
  15. */ 
    16.   
  
  
  
  16. public PriorityBlockingQueue(int initialCapacity, 
    17.   
  
  
  
  17.                              Comparator comparator) { 
    18.   
  
  
  
  18.     if (initialCapacity < 1) 
    19.   
  
  
  
  19.         throw new IllegalArgumentException(); 
    20.   
  
  
  
  20.     this.lock = new ReentrantLock(); 
    21.   
  
  
  
  21.     this.notEmpty = lock.newCondition(); 
    22.   
  
  
  
  22.     this.comparator = comparator; 
    23.   
  
  
  
  23.     this.queue = new Object[initialCapacity]; 
    24.   
  
  
  
  24. } 
    25.   
  
  
  
  25. /** 
    26.   
  
  
  
  26. * 構(gòu)造函數(shù)，帶有初始內(nèi)容的隊(duì)列 
    27.   
  
  
  
  27. */ 
    28.   
  
  
  
  28. public PriorityBlockingQueue(Collection c) { 
    29.   
  
  
  
  29.     this.lock = new ReentrantLock(); 
    30.   
  
  
  
  30.     this.notEmpty = lock.newCondition(); 
    31.   
  
  
  
  31.     boolean heapify = true; // true if not known to be in heap order 
    32.   
  
  
  
  32.     boolean screen = true;  // true if must screen for nulls 
    33.   
  
  
  
  33.     if (c instanceof SortedSet) { 
    34.   
  
  
  
  34.         SortedSet ss = (SortedSet) c; 
    35.   
  
  
  
  35.         this.comparator = (Comparator) ss.comparator(); 
    36.   
  
  
  
  36.         heapify = false; 
    37.   
  
  
  
  37.     } 
    38.   
  
  
  
  38.     else if (c instanceof PriorityBlockingQueue) { 
    39.   
  
  
  
  39.         PriorityBlockingQueue pq = 
    40.   
  
  
  
  40.             (PriorityBlockingQueue) c; 
    41.   
  
  
  
  41.         this.comparator = (Comparator) pq.comparator(); 
    42.   
  
  
  
  42.         screen = false; 
    43.   
  
  
  
  43.         if (pq.getClass() == PriorityBlockingQueue.class) // exact match 
    44.   
  
  
  
  44.             heapify = false; 
    45.   
  
  
  
  45.     } 
    46.   
  
  
  
  46.     Object[] a = c.toArray(); 
    47.   
  
  
  
  47.     int n = a.length; 
    48.   
  
  
  
  48.     // If c.toArray incorrectly doesn't return Object[], copy it. 
    49.   
  
  
  
  49.     if (a.getClass() != Object[].class) 
    50.   
  
  
  
  50.         a = Arrays.copyOf(a, n, Object[].class); 
    51.   
  
  
  
  51.     if (screen && (n == 1 || this.comparator != null)) { 
    52.   
  
  
  
  52.         for (int i = 0; i < n; ++i) 
    53.   
  
  
  
  53.             if (a[i] == null) 
    54.   
  
  
  
  54.                 throw new NullPointerException(); 
    55.   
  
  
  
  55.     } 
    56.   
  
  
  
  56.     this.queue = a; 
    57.   
  
  
  
  57.     this.size = n; 
    58.   
  
  
  
  58.     if (heapify) 
    59.   
  
  
  
  59.         heapify(); 
    60.   
  
  
  
  60. } 
    61.

 入隊(duì)方法

入隊(duì)方法，下面可以看到 put 方法最終會調(diào)用 offer 方法，所以我們只看 offer 方法即可。

offer(E e)

 
 
 
 
      
  
  
  
  1. public void put(E e) { 
    2.   
  
  
  
  2.     offer(e); // never need to block 
    3.   
  
  
  
  3. } 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5. public boolean offer(E e) { 
    6.   
  
  
  
  6.     //判斷是否為空 
    7.   
  
  
  
  7.     if (e == null) 
    8.   
  
  
  
  8.         throw new NullPointerException(); 
    9.   
  
  
  
  9.     //顯示鎖 
    10.   
  
  
  
  10.     final ReentrantLock lock = this.lock; 
    11.   
  
  
  
  11.     lock.lock(); 
    12.   
  
  
  
  12.     //定義臨時(shí)對象 
    13.   
  
  
  
  13.     int n, cap; 
    14.   
  
  
  
  14.     Object[] array; 
    15.   
  
  
  
  15.     //判斷數(shù)組是否滿了 
    16.   
  
  
  
  16.     while ((n = size) >= (cap = (array = queue).length)) 
    17.   
  
  
  
  17.         //數(shù)組擴(kuò)容 
    18.   
  
  
  
  18.         tryGrow(array, cap); 
    19.   
  
  
  
  19.     try { 
    20.   
  
  
  
  20.         //拿到比較器 
    21.   
  
  
  
  21.         Comparator cmp = comparator; 
    22.   
  
  
  
  22.         //判斷是否有自定義比較器 
    23.   
  
  
  
  23.         if (cmp == null) 
    24.   
  
  
  
  24.             //堆上浮 
    25.   
  
  
  
  25.             siftUpComparable(n, e, array); 
    26.   
  
  
  
  26.         else 
    27.   
  
  
  
  27.             //使用自定義比較器進(jìn)行堆上浮 
    28.   
  
  
  
  28.             siftUpUsingComparator(n, e, array, cmp); 
    29.   
  
  
  
  29.         //隊(duì)列長度 +1 
    30.   
  
  
  
  30.         size = n + 1; 
    31.   
  
  
  
  31.         //喚醒休眠的出隊(duì)線程 
    32.   
  
  
  
  32.         notEmpty.signal(); 
    33.   
  
  
  
  33.     } finally { 
    34.   
  
  
  
  34.         //釋放鎖 
    35.   
  
  
  
  35.         lock.unlock(); 
    36.   
  
  
  
  36.     } 
    37.   
  
  
  
  37.     return true; 
    38.   
  
  
  
  38. } 
    39.

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

上浮調(diào)整比較器方法的實(shí)現(xiàn)

 
 
 
 
      
  
  
  
  1. private static  void siftUpComparable(int k, T x, Object[] array) { 
    2.   
  
  
  
  2.         Comparable key = (Comparable) x; 
    3.   
  
  
  
  3.         while (k > 0) { 
    4.   
  
  
  
  4.          //無符號向左移，目的是找到放入位置的父節(jié)點(diǎn) 
    5.   
  
  
  
  5.             int parent = (k - 1) >>> 1; 
    6.   
  
  
  
  6.             //拿到父節(jié)點(diǎn)的值 
    7.   
  
  
  
  7.             Object e = array[parent]; 
    8.   
  
  
  
  8.             //比較是否大于該元素，不大于就沒比較交換 
    9.   
  
  
  
  9.             if (key.compareTo((T) e) >= 0) 
    10.   
  
  
  
  10.                 break; 
    11.   
  
  
  
  11.             //以下都是元素位置交換 
    12.   
  
  
  
  12.             array[k] = e; 
    13.   
  
  
  
  13.             k = parent; 
    14.   
  
  
  
  14.         } 
    15.   
  
  
  
  15.         array[k] = key; 
    16.   
  
  
  
  16.     } 
    17.

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

入隊(duì)圖解

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

 
 
 
 
      
  
  
  
  1. package com.niuh.queue.priority; 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import java.util.Comparator; 
    4.   
  
  
  
  4. import java.util.concurrent.PriorityBlockingQueue; 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6. /** 
    7.   
  
  
  
  7.  * 
     
    8.   
  
  
  
  8.  * PriorityBlockingQueue 簡單演示 demo 
    9.   
  
  
  
  9.  * 
     
    10.   
  
  
  
  10.  */ 
    11.   
  
  
  
  11. public class TestPriorityBlockingQueue { 
    12.   
  
  
  
  12.  
    13.   
  
  
  
  13.     public static void main(String[] args) throws InterruptedException { 
    14.   
  
  
  
  14.         // 大頂堆 
    15.   
  
  
  
  15.         PriorityBlockingQueue concurrentLinkedQueue = new PriorityBlockingQueue(10, new Comparator() { 
    16.   
  
  
  
  16.             @Override 
    17.   
  
  
  
  17.             public int compare(Integer o1, Integer o2) { 
    18.   
  
  
  
  18.                 return o2 - o1; 
    19.   
  
  
  
  19.             } 
    20.   
  
  
  
  20.         }); 
    21.   
  
  
  
  21.  
    22.   
  
  
  
  22.         concurrentLinkedQueue.offer(90); 
    23.   
  
  
  
  23.         concurrentLinkedQueue.offer(80); 
    24.   
  
  
  
  24.         concurrentLinkedQueue.offer(70); 
    25.   
  
  
  
  25.         concurrentLinkedQueue.offer(60); 
    26.   
  
  
  
  26.         concurrentLinkedQueue.offer(40); 
    27.   
  
  
  
  27.         concurrentLinkedQueue.offer(30); 
    28.   
  
  
  
  28.         concurrentLinkedQueue.offer(20); 
    29.   
  
  
  
  29.         concurrentLinkedQueue.offer(10); 
    30.   
  
  
  
  30.         concurrentLinkedQueue.offer(50); 
    31.   
  
  
  
  31.         concurrentLinkedQueue.offer(85); 
    32.   
  
  
  
  32.         //輸出元素排列 
    33.   
  
  
  
  33.         concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  ")); 
    34.   
  
  
  
  34.         //取出元素 
    35.   
  
  
  
  35.         Integer take = concurrentLinkedQueue.take(); 
    36.   
  
  
  
  36.         System.out.println(); 
    37.   
  
  
  
  37.         concurrentLinkedQueue.stream().forEach(e-> System.out.print(e+"  ")); 
    38.   
  
  
  
  38.     } 
    39.   
  
  
  
  39. } 
    40.

操作的細(xì)節(jié)分為兩步：

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

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

85 再繼續(xù)往前走之后，它要和 80 再進(jìn)行比較，同理可得：也就是說這個(gè)結(jié)點(diǎn)每次和它的父親比，如果它大于它的父親的話就交換，直到它不再大于它的父親。

出隊(duì)方法

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

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

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

take()

出隊(duì)方法，該方法會阻塞

 
 
 
 
      
  
  
  
  1. public E take() throws InterruptedException { 
    2.   
  
  
  
  2.  //顯示鎖 
    3.   
  
  
  
  3.     final ReentrantLock lock = this.lock; 
    4.   
  
  
  
  4.     //可中斷鎖 
    5.   
  
  
  
  5.     lock.lockInterruptibly(); 
    6.   
  
  
  
  6.     //結(jié)果接收對象 
    7.   
  
  
  
  7.     E result; 
    8.   
  
  
  
  8.     try { 
    9.   
  
  
  
  9.      //判斷隊(duì)列是否為空 
    10.   
  
  
  
  10.         while ( (result = dequeue()) == null) 
    11.   
  
  
  
  11.          //線程阻塞 
    12.   
  
  
  
  12.             notEmpty.await(); 
    13.   
  
  
  
  13.     } finally { 
    14.   
  
  
  
  14.         lock.unlock(); 
    15.   
  
  
  
  15.     } 
    16.   
  
  
  
  16.     return result; 
    17.   
  
  
  
  17. } 
    18.

 dequeue()

我們再來看看具體出隊(duì)方法的實(shí)現(xiàn)，dequeue方法

 
 
 
 
      
  
  
  
  1. private E dequeue() { 
    2.   
  
  
  
  2. //長度減少 1 
    3.   
  
  
  
  3.    int n = size - 1; 
    4.   
  
  
  
  4.    //判斷隊(duì)列中是否有元素 
    5.   
  
  
  
  5.    if (n < 0) 
    6.   
  
  
  
  6.        return null; 
    7.   
  
  
  
  7.    else { 
    8.   
  
  
  
  8.     //隊(duì)列對象 
    9.   
  
  
  
  9.        Object[] array = queue; 
    10.   
  
  
  
  10.        //取出第一個(gè)元素 
    11.   
  
  
  
  11.        E result = (E) array[0]; 
    12.   
  
  
  
  12.        //拿出最后一個(gè)元素 
    13.   
  
  
  
  13.        E x = (E) array[n]; 
    14.   
  
  
  
  14.        //置空 
    15.   
  
  
  
  15.        array[n] = null; 
    16.   
  
  
  
  16.        Comparator cmp = comparator; 
    17.   
  
  
  
  17.        if (cmp == null) 
    18.   
  
  
  
  18.         //下沉調(diào)整 
    19.   
  
  
  
  19.            siftDownComparable(0, x, array, n); 
    20.   
  
  
  
  20.        else 
    21.   
  
  
  
  21.            siftDownUsingComparator(0, x, array, n, cmp); 
    22.   
  
  
  
  22.        //成功則減少隊(duì)列中的元素?cái)?shù)量 
    23.   
  
  
  
  23.        size = n; 
    24.   
  
  
  
  24.        return result; 
    25.   
  
  
  
  25.    } 
    26.

 總體就是找到父節(jié)點(diǎn)與兩個(gè)子節(jié)點(diǎn)中最小的一個(gè)節(jié)點(diǎn)，然后進(jìn)行交換位置，不斷重復(fù)，由上而下的交換。

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

再來看看下沉比較器方法的實(shí)現(xiàn)

 
 
 
 
      
  
  
  
  1. private static  void siftDownComparable(int k, T x, Object[] array, 
    2.   
  
  
  
  2.                                                int n) { 
    3.   
  
  
  
  3.     //判斷隊(duì)列長度 
    4.   
  
  
  
  4.     if (n > 0) { 
    5.   
  
  
  
  5.         Comparable key = (Comparable)x; 
    6.   
  
  
  
  6.         //找到隊(duì)列最后一個(gè)元素的父節(jié)點(diǎn)的索引。 
    7.   
  
  
  
  7.         int half = n >>> 1;           // loop while a non-leaf 
    8.   
  
  
  
  8.         while (k < half) { 
    9.   
  
  
  
  9.          //拿到 k 節(jié)點(diǎn)下的左子節(jié)點(diǎn) 
    10.   
  
  
  
  10.             int child = (k << 1) + 1; // assume left child is least 
    11.   
  
  
  
  11.             //取得子節(jié)點(diǎn)對應(yīng)的值 
    12.   
  
  
  
  12.             Object c = array[child]; 
    13.   
  
  
  
  13.             //取得 k 右子節(jié)點(diǎn)的索引 
    14.   
  
  
  
  14.             int right = child + 1; 
    15.   
  
  
  
  15.             //比較右節(jié)點(diǎn)的索引是否小于隊(duì)列長度和左右子節(jié)點(diǎn)的值進(jìn)行比較 
    16.   
  
  
  
  16.             if (right < n && 
    17.   
  
  
  
  17.                 ((Comparable) c).compareTo((T) array[right]) > 0) 
    18.   
  
  
  
  18.                 c = array[child = right]; 
    19.   
  
  
  
  19.             //比較父節(jié)點(diǎn)值是否大于子節(jié)點(diǎn) 
    20.   
  
  
  
  20.             if (key.compareTo((T) c) <= 0) 
    21.   
  
  
  
  21.                 break; 
    22.   
  
  
  
  22.             //下面都是元素替換 
    23.   
  
  
  
  23.             array[k] = c; 
    24.   
  
  
  
  24.             k = child; 
    25.   
  
  
  
  25.         } 
    26.   
  
  
  
  26.         array[k] = key; 
    27.   
  
  
  
  27.     } 
    28.   
  
  
  
  28. } 
    29.

 出隊(duì)圖解

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

依次從根部向下調(diào)整整個(gè)堆的結(jié)構(gòu)(一直到堆尾即可) HeapifyDown

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

總結(jié)

PriorityBlockingQueue 真的是個(gè)神奇的隊(duì)列，可以實(shí)現(xiàn)優(yōu)先出隊(duì)。最特別的是它只有一個(gè)鎖，入隊(duì)操作永遠(yuǎn)成功，而出隊(duì)只有在空隊(duì)列的時(shí)候才會進(jìn)行線程阻塞?？梢哉f有一定的應(yīng)用場景吧，比如：有任務(wù)要執(zhí)行，可以對任務(wù)加一個(gè)優(yōu)先級的權(quán)重，這樣隊(duì)列會識別出來，對該任務(wù)優(yōu)先進(jìn)行出隊(duì)。


當(dāng)前名稱：阻塞隊(duì)列—PriorityBlockingQueue源碼分析
URL標(biāo)題：http://uogjgqi.cn/article/dhoejgg.html
掃二維碼與項(xiàng)目經(jīng)理溝通

我們在微信上24小時(shí)期待你的聲音

解答本文疑問/技術(shù)咨詢/運(yùn)營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流