併發Queue(ConcurrentLinkedQueue和BlockingQueue)
在併發佇列上JDK提供了兩套實現,一個是以ConcurrentLinkedQueue為代表的高效能佇列,一個是以BlockingQueue介面為代表的阻塞佇列,無論哪種都繼承自Queue
一、ConCurrentLinkedQueue:
是一個適用於高併發場景下的佇列,通過無鎖的方式,實現了高併發狀態下的高效能,通常ConcurrentLinkedQueue效能好於BlockingQueue。它是一個基於連結節點的無界執行緒安全佇列。該佇列的元素遵循先進先出的原則。頭是最先加入的,尾是最近加入的,該佇列不允許null元素。
ConcurrentLinkedQueue重要方法:
add()和offer()都是加入元素的節點(在ConcurrentLinkedQueue中,這兩個方法沒有任何區別)
poll()和peek()都是取頭元素節點,區別在於前者會刪除元素,後者不會。
示例:
public static void main(String[] args) {
//高效能無阻塞無界佇列
ConcurrentLinkedQueue<String> cl=new ConcurrentLinkedQueue<>();
cl.add("a");
cl.offer("b");
cl.add("c");
cl.offer("d");
cl.add("e");
System.out.println("佇列原始大小:"+cl.size());
System.out.println(cl.poll());
System.out.println("poll取值後的大小:"+cl.size());
System.out.println(cl.peek());
System.out.println("peek取值後的大小:"+cl.size());
}
輸出結果為:
佇列原始大小:5
a
poll取值後的大小:4
b
peek取值後的大小:4
二、BlockingQueue介面
常用方法
丟擲異常 | 特殊值 | 阻塞 | 超時 | |
插入 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除 | remove() | poll() | take() | poll(time,unit) |
檢查 | element() | peek() | 不可用 | 不可用 |
- ArrayBlockingQueue:基於陣列的阻塞佇列實現,在ArrayBlockingQueue內部,維護了一個定長陣列,以便快取佇列中的資料物件,其內部沒實現讀寫分離,也就意味著生成和消費不能完全並行,長度是需要定義的,可以指定先進先出或者先進後出,也叫有界佇列,在很多場合非常適合使用。
- LinkedBlockingQueue:基於連結串列的阻塞佇列,同ArrayBlockingQueue類似,其內部也維持著一個數據緩衝佇列(該佇列由一個連結串列構成),LinkedBlockingQueue之所以能夠高效的處理併發資料,是因為其內部實現採用分離鎖(讀寫分離兩個鎖),從而實現生產者和消費者操作的完全並行執行。他是一個無界佇列。
- SynchronousQueue:一種沒有緩衝的佇列,生產者產出的資料直接會被消費者獲取並消費。(不可以新增元素)
- PriorityBlockingQueue:基於優先順序的阻塞佇列(優先順序的判斷通過建構函式傳入的Compator物件來決定,也就是說傳入佇列的物件必須實現Comparable介面),在實現PriortyBlockingQueue時,內部控制執行緒同步的鎖採用的是公平鎖,他也是一個無界的佇列。每次出隊都返回優先順序最高的元素,是二叉樹最小堆的實現,研究過陣列方式存放最小堆節點的都知道,直接遍歷佇列元素是無序的。
PriorityBlockingQueue 排序時機劃分為兩處
插入 - put/add/offer 對應使用 siftUpComparable 排序方法;
每當加入一個元素 , 與當前佇列隊首元素compareTo比較 , 根據返回值決定是否排到隊首
彈出 - poll(remove)/take 對應使用 siftDownComparable 排序方法;
呼叫 dequeue 方法直接彈出隊首元素 , 由於插入時保證隊首永遠為 compareTo 為1 的元素即符合判斷順序的元素 , siftDownComparable 方法使用建立最小堆的演算法做元素重排序 , 排序細節參考http://www.sohu.com/a/165802088_355142- DelayQueue:帶有延遲時間的Queue,其中的元素只有當其指定的延遲時間到了,才能夠從佇列中獲取到該元素。DelayQueue中的元素必須實現Delayed介面,DelayQueue是一個沒有大小限制的佇列,應用場景很多,比如對快取超時的資料進行移出、任務超時處理、空閒連結的關閉等等。
ArrayBlockingQueue示例:
ArrayBlockingQueue<String> array=new ArrayBlockingQueue<>(5);
array.put("a");
array.put("b");
array.put("c");
array.add("d");
array.add("e");
//put()當元素達到佇列的最大值當前阻塞
//add()會拋異常java.lang.IllegalStateException: Queue full
// array.add("f");
System.out.println(array.offer("a", 2, TimeUnit.SECONDS));//兩秒過後返回true:新增成功,false:新增失敗
for (String string : array) {
System.out.println(string);
}
輸出結果:
false
a
b
c
d
e
LinkedBlockingQueue示例:
//無界阻塞佇列,初始化可以不給定長度(表示無界),也可以給定長度(表示有界只能新增給定長度的元素)
LinkedBlockingQueue<String> linked=new LinkedBlockingQueue<>();
linked.put("a");
linked.put("b");
linked.put("c");
linked.add("d");
linked.add("e");
linked.offer("f");
System.out.println(linked.offer("a", 2, TimeUnit.SECONDS));//兩秒過後返回true:新增成功,false:新增失敗
for (String string : linked) {
System.out.println(string);
}System.out.println("----------------");
List<String> list=new ArrayList<>();
linked.drainTo(list, 3);//表示從佇列中獲取3個元素放入集合中
System.out.println("list.size="+list.size());
for (String str : list) {
System.out.println(str);
}
輸出結果:
true
a
b
c
d
e
f
a----------------
list.size=3
a
b
c
SynchronousQueue示例:必須先take()阻塞在add()加入元素,但是不是加入元素到容器而是直接丟給take()的執行緒
SynchronousQueue<String> q=new SynchronousQueue<>();
Thread t1=new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println(q.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"t1");
t1.start();
Thread t2=new Thread(new Runnable() {
@Override
public void run() {
q.add("abcd");
}
},"t2");
t2.start();
PriorityBlockingQueue示例:
Task類:
public class Task implements Comparable<Task>{
private int id;
private String name;
public Task(int id, String name) {
this.id = id;
this.name = name;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Task [id=" + id + ", name=" + name + "]";
}
@Override
public int compareTo(Task task) {
return this.id > task.id ? 1 : (this.id < task.id ? -1 : 0);
}}
Main方法:
public class UsePriorityBlockingQueue {
public static void main(String[] args) throws InterruptedException {
PriorityBlockingQueue<Task> p=new PriorityBlockingQueue<Task>();
Task t1=new Task(5,"任務1");
Task t2=new Task(4,"任務2");
Task t3=new Task(3,"任務3");
Task t4=new Task(6,"任務4");
p.put(t1);
System.out.println(p);
p.put(t2);
System.out.println(p);
p.put(t3);
System.out.println(p);
p.put(t4);
System.out.println(p);
System.out.println("------------------------------");
for (Task task : p) {
System.out.println(task);
}
System.out.println("------------------------------");
System.out.println(p.poll());
System.out.println(p.poll());
System.out.println(p.poll());
System.out.println(p.poll());
}}
輸出結果:
[Task [id=5, name=任務1]]
[Task [id=4, name=任務2], Task [id=5, name=任務1]]
[Task [id=3, name=任務3], Task [id=5, name=任務1], Task [id=4, name=任務2]]
[Task [id=3, name=任務3], Task [id=5, name=任務1], Task [id=4, name=任務2], Task [id=6, name=任務4]]
------------------------------
Task [id=3, name=任務3]
Task [id=5, name=任務1]
Task [id=4, name=任務2]
Task [id=6, name=任務4]
------------------------------
Task [id=3, name=任務3]
Task [id=4, name=任務2]
Task [id=5, name=任務1]
Task [id=6, name=任務4]
DelayQueue示例:
Wangmin類:
/**
* 網名
* @author Administrator
*
*/
public class Wangmin implements Delayed{
private String name;
//身份證
private String id;
//截止時間
private long endTime;
//定義時間工具類
private TimeUnit timeUnit=TimeUnit.SECONDS;
public Wangmin(String name, String id, long endTime) {
this.name = name;
this.id = id;
this.endTime = endTime;
}public String getName() {
return name;
}public void setName(String name) {
this.name = name;
}public String getId() {
return id;
}public void setId(String id) {
this.id = id;
}public long getEndTime() {
return endTime;
}public void setEndTime(long endTime) {
this.endTime = endTime;
}/**
* 相互比較排序用
*/
@Override
public int compareTo(Delayed o) {
Wangmin w=(Wangmin)o;
return this.getDelay(timeUnit)-w.getDelay(timeUnit)>0?1:0;
}/**
* 判斷是否到了截止時間
*/
@Override
public long getDelay(TimeUnit unit) {
return endTime-System.currentTimeMillis();
}}
Wangba類:
public class Wangba implements Runnable{
//用於存放網名的佇列
private DelayQueue<Wangmin> queue=new DelayQueue<>();
//營業標識
public boolean yingye=true;
/**
* 上機方法
* @param name
* @param id
* @param money
*/
public void shangji(String name,String id,int money) {
Wangmin wm=new Wangmin(name, id, 1000*money+System.currentTimeMillis());
System.out.println("網名"+wm.getName()+"身份證"+wm.getId()+"交錢"+money+"元,開始上機...");
this.queue.put(wm);
}
public void xiaji(Wangmin wm) {
System.out.println("網名"+wm.getName()+"身份證"+wm.getId()+"時間到下機");
}
@Override
public void run() {
while (yingye) {
try {
Wangmin wm=queue.take();
xiaji(wm);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}public static void main(String[] args) {
System.out.println("開始營業");
Wangba wb=new Wangba();
Thread shangwang=new Thread(wb);
shangwang.start();
wb.shangji("路人甲", "123", 1);
wb.shangji("路人乙", "234", 5);
wb.shangji("路人丙", "345", 3);
}
}
輸出結果:
開始營業
網名路人甲身份證123交錢1元,開始上機...
網名路人乙身份證234交錢5元,開始上機...
網名路人丙身份證345交錢3元,開始上機...
網名路人甲身份證123時間到下機
網名路人丙身份證345時間到下機
網名路人乙身份證234時間到下機