在并发队列上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时间到下机