併發程式設計--Queue

Queue

• - 非阻塞佇列
• - 阻塞佇列

Queue是一種特殊的線性表，它只允許在表的前端（front）進行刪除操作，而在表的後端（rear）進行插入操作。進行插入操作的端稱為隊尾，進行刪除操作的端稱為隊頭。佇列中沒有元素時，稱為空佇列。
在佇列這種資料結構中，最先插入的元素將是最先被刪除的元素；反之最後插入的元素將是最後被刪除的元素，因此佇列又稱為“先進先出”（FIFO—first in first out）的線性表。
在java5中新增加了java.util.Queue介面，用以支援佇列的常見操作。該介面擴充套件了java.util.Collection介面。主要提供兩類的實現：

• * ConcurrentLinkedQueue為代表的高效能的佇列
• * BlockingQueue 為代表的阻塞佇列

1-1 非阻塞佇列

ConcurrentLinkedQueue
ConcurrentLinkedQueue是一個基於連結節點的無界執行緒安全佇列，它採用先進先出的規則對節點進行排序，當我們新增一個元素的時候，它會新增到佇列的尾部，當我們獲取一個元素時，它會返回佇列頭部的元素。
它採用了“wait－free”演算法來實現，該演算法在Michael & Scott演算法上進行了一些修改, Michael & Scott演算法的詳細資訊可以參見參考資料一 http://www.cs.rochester.edu/u/michael/PODC96.html
注意他的幾個方法：
add(E e) 和 offer(E e) ： 都是加入元素的方法 （ConcurrentLinkedQueue 這兩個方法沒有區別）
poll() 和 peek() 都是取頭元素 ， 區別在於poll() 獲取並移除此佇列的頭，如果此佇列為空，則返回 null。peek()不會刪除。

 1 public class MyQueue {
 2 
 3   public static void main(String[] args) throws Exception {
 4 
 5     //高效能無阻塞無界佇列：ConcurrentLinkedQueue
 6 
 7     final SynchronousQueue<String> q = new SynchronousQueue<String>();
 8     Thread t1 = new Thread(new Runnable() {
 9       @Override
10       public
 
 void run() {
11         try {
12           System.out.println(q.take());
13         } catch (InterruptedException e) {
14           e.printStackTrace();
15         }
16       }
17     });
18     t1.start();
19     Thread t2 = new Thread(new Runnable() {
20 
21       @Override
22       public void run() {
23         q.add("asdasd");
24       }
25     });
26     t2.start();        
27   }
28 }

ConcurrentLinkedDeque

 1 import java.util.concurrent.LinkedBlockingDeque;
 2 
 3 public class MyDeque {
 4 
 5   public static void main(String[] args) {
 6 
 7     LinkedBlockingDeque<String> dq = new LinkedBlockingDeque<String>(10);
 8     dq.addFirst("a");
 9     dq.addFirst("b");
10     dq.addFirst("c");
11     dq.addFirst("d");
12     dq.addFirst("e");
13     dq.addLast("f");
14     dq.addLast("g");
15     dq.addLast("h");
16     dq.addLast("i");
17     dq.addLast("j");
18     //dq.offerFirst("k");
19     System.out.println("檢視頭元素：" + dq.peekFirst());
20     System.out.println("獲取尾元素：" + dq.pollLast());
21     Object [] objs = dq.toArray();
22     for (int i = 0; i < objs.length; i++) {
23       System.out.println(objs[i]);
24     }
25 
26   }
27 }

1-2 阻塞佇列

JDK提供的阻塞佇列：

• ArrayBlockingQueue：

一個由陣列結構組成的 有界阻塞佇列，遵循FIFO原則。其內部維護一個定長的陣列，以便快取佇列中的資料物件，其內部沒有實現讀寫分離，生產者和消費者不能完全並行。
ArrayBlockingQueue是一個用陣列實現的有界阻塞佇列。此佇列按照先進先出（FIFO）的原則對元素進行排序。預設情況下不保證訪問者公平的訪問佇列，所謂公平訪問佇列是指阻塞的所有生產者執行緒或消費者執行緒，
當佇列可用時，可以按照阻塞的先後順序訪問佇列，即先阻塞的生產者執行緒，可以先往佇列裡插入元素，先阻塞的消費者執行緒，可以先從佇列裡獲取元素。通常情況下為了保證公平性會降低吞吐量。
我們可以使用以下程式碼建立一個公平的阻塞佇列：
ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
訪問者的公平性是使用可重入鎖實現的，程式碼如下：

1 public ArrayBlockingQueue(int capacity, boolean fair) {
2     if (capacity <= 0)
3         throw new IllegalArgumentException();
4     this.items = new Object[capacity];
5     lock = new ReentrantLock(fair);
6     notEmpty = lock.newCondition();
7     notFull =  lock.newCondition();
8 }

• LinkedBlockingQueue：

一個由連結串列結構組成的有界阻塞佇列，遵循FIFO原則，預設和最大長度為Integer.MAX_VALUE。同ArrayBlockingQueue類似，其內部也是維護一個數據緩衝佇列（由連結串列構成），
LinkedBlockingQueue 之所以可以高效的處理併發資料，是因為LinkedBlockingQueue內部採用分離鎖（讀寫分離兩個鎖），從而實現生產者和消費者的操作完成並行執行。

• PriorityBlockingQueue：

一個支援優先順序排序的無界阻塞佇列。其排序不是加入的時候完成 而是呼叫take()方法時候進行排序
預設情況下元素採取自然順序排列，也可以通過比較器comparator來指定元素的排序規則。元素按照升序排列。
實現PriorityBlockingQueue時，內部控制執行緒同步的鎖採用的是公平鎖。

示例：MyPriorityBlockingQueue.Java

 1 public class Task implements Comparable<Task>{
 2     private int id ;
 3     private String name;
 4     public int getId() {
 5       return id;
 6     }
 7     public void setId(int id) {
 8       this.id = id;
 9     }
10     public String getName() {
11       return name;
12     }
13     public void setName(String name) {
14       this.name = name;
15     }
16 
17     @Override
18     public int compareTo(Task task) {
19       return this.id > task.id ? 1 : (this.id < task.id ? -1 : 0);  
20     }
21 
22     public String toString(){
23       return this.id + "," + this.name;
24     }
25 }

 1 import java.util.ArrayList;
 2 import java.util.Collections;
 3 import java.util.Iterator;
 4 import java.util.List;
 5 import java.util.concurrent.PriorityBlockingQueue;
 6 
 7 public class MyPriorityBlockingQueue {
 8     public static void main(String[] args) throws Exception{
 9       PriorityBlockingQueue<Task> q = new PriorityBlockingQueue<Task>();
10 
11       Task t1 = new Task();
12       t1.setId(3);
13       t1.setName("id為3");
14       Task t2 = new Task();
15       t2.setId(4);
16       t2.setName("id為4");
17       Task t3 = new Task();
18       t3.setId(1);
19       t3.setName("id為1");
20 
21       //return this.id > task.id ? 1 : 0;
22       q.add(t1);    //3
23       q.add(t2);    //4
24       q.add(t3);  //1
25 
26       // 1 3 4
27       System.out.println("容器：" + q);
28       System.out.println(q.take().getId()); //其排序不是加入的時候完成 而是呼叫take()方法時候進行排序
29       System.out.println("容器：" + q);
30     //        System.out.println(q.take().getId());
31     //        System.out.println(q.take().getId());
32     }
33 }

結果輸出

容器：[1,id為1, 4,id為4, 3,id為3]
1
容器：[3,id為3, 4,id為4]   

• DelayQueue：一個使用優先順序佇列實現的無界阻塞佇列。

佇列使用PriorityQueue來實現。佇列中的元素必須實現Delayed介面(佇列中的Delayed必須實現compareTo來指定元素的順序。比如讓延時時間最長的放在佇列的末尾)，在建立元素時可以指定多久才能從佇列中獲取當前元素。只有在延遲期滿時才能從佇列中提取元素。
我們可以將DelayQueue運用在以下應用場景：
* 快取系統的設計：可以用DelayQueue儲存快取元素的有效期，使用一個執行緒迴圈查詢DelayQueue，一旦能從DelayQueue中獲取元素時，表示快取有效期到了。
* 定時任務排程。使用DelayQueue儲存當天將會執行的任務和執行時間，一旦從DelayQueue中獲取到任務就開始執行，從比如TimerQueue就是使用DelayQueue實現的。

• SynchronousQueue：一個不儲存元素的阻塞佇列。

SynchronousQueue是一個不儲存元素的阻塞佇列。每一個put操作必須等待一個take操作，否則不能繼續新增元素。SynchronousQueue可以看成是一個傳球手，負責把生產者執行緒處理的資料直接傳遞給消費者執行緒。
佇列本身並不儲存任何元素，非常適合於傳遞性場景,比如在一個執行緒中使用的資料，傳遞給另外一個執行緒使用，SynchronousQueue的吞吐量高於LinkedBlockingQueue 和 ArrayBlockingQueue。

• LinkedTransferQueue：一個由連結串列結構組成的無界阻塞佇列。

LinkedTransferQueue是一個由連結串列結構組成的無界阻塞TransferQueue佇列。相對於其他阻塞佇列LinkedTransferQueue多了tryTransfer和transfer方法。
* transfer方法。如果當前有消費者正在等待接收元素（消費者使用take()方法或帶時間限制的poll()方法時），transfer方法可以把生產者傳入的元素立刻transfer（傳輸）給消費者。
如果沒有消費者在等待接收元素，transfer方法會將元素存放在佇列的tail節點，並等到該元素被消費者消費了才返回。

transfer方法的關鍵程式碼如下：

Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

第一行程式碼是試圖把存放當前元素的s節點作為tail節點。
第二行程式碼是讓CPU自旋等待消費者消費元素。因為自旋會消耗CPU，所以自旋一定的次數後使用Thread.yield()方法來暫停當前正在執行的執行緒，並執行其他執行緒。

• * tryTransfer方法。則是用來試探下生產者傳入的元素是否能直接傳給消費者。如果沒有消費者等待接收元素，則返回false。

和transfer方法的區別是: tryTransfer方法無論消費者是否接收，方法立即返回。而transfer方法是必須等到消費者消費了才返回。
對於帶有時間限制的tryTransfer(E e, long timeout, TimeUnit unit)方法，則是試圖把生產者傳入的元素直接傳給消費者，但是如果沒有消費者消費該元素則等待指定的時間再返回，
如果超時還沒消費元素，則返回false，如果在超時時間內消費了元素，則返回true。

LinkedBlockingDeque是一個由連結串列結構組成的雙向阻塞佇列。所謂雙向佇列指的你可以從佇列的兩端插入和移出元素。雙端佇列因為多了一個操作佇列的入口，在多執行緒同時入隊時，也就減少了一半的競爭。
相比其他的阻塞佇列，LinkedBlockingDeque多了addFirst，addLast，offerFirst，offerLast，peekFirst，peekLast等方法，以First單詞結尾的方法，表示插入，獲取（peek）或移除雙端佇列的第一個元素。
以Last單詞結尾的方法，表示插入，獲取或移除雙端佇列的最後一個元素。另外插入方法add等同於addLast，移除方法remove等效於removeFirst。
但是take方法卻等同於takeFirst，不知道是不是Jdk的bug，使用時還是用帶有First和Last字尾的方法更清楚。在初始化LinkedBlockingDeque時可以初始化佇列的容量，用來防止其再擴容時過渡膨脹。
另外雙向阻塞佇列可以運用在“工作竊取”模式中。