Queue的理解

并发Queue

在哪并发队列上jdk提供了两套实现,一个是以ConcurrentLinkedQueue为代表的高性能队列,一个是以BlockingQueue为代表的阻塞队列,无论哪种都继承自Queue

1.ConcurrentLinkedQueue

是一个使用与高并发场景下的队列,通过无锁的方式,实现类高并发状态下的高性能,通常ConcurrentLinkedQueue性能好于BlockingQueue.它是基于链接点无界 线程安全队列.该队列的元素遵循先进先出的原则.头是加入的,尾是最近加入的,该队列不允许null值 这个队列使用链表作为其数据结构。这个类算是高并发环境中性能最好的队列。高性能是因为其内部复杂的实现。

ConcurrentLinkedQueue的重要方法:

add()和offer()都是加入元素的方法(在ConcurrentLinkedQueue,这两个方法没有任何区别) poll()和peak()都是取头元素,区别在于前者会删除元素后者不会.

2.BlockingQueue接口

ArrayBlockingQueue:

基于数组的阻塞队列实现,在ArrayBlockingQueue内部,维护了一个定长的数组,一遍缓存队列中的数据对象,对内部没实现读写分离,也就意味着生产和消费 不能完全并行,长度是需要定义的,可以指定现进先出,或者,后进后出,也叫有界队列,在很多场合非常实用,

LinkenBlockingQueue:

基于链表的阻塞队列,同ArrayBlockingQueue类似,器内部也维护这一个数据缓存队列(该队列由一个链表构成),LinkedBlockingQueue之所以能够高效地 处理并发数据,是因为其内部实现采用分离锁(读写分离两个锁),从而实现生产者和消费者操作的完全并行运行.它是一个无界队列

SynchronousQueue:

一种没有缓冲的队列,生产者产生的数据直接会被消费者获取并消费.

看一个具体的例子:

SynchronousQueue<String> sQueue = new SynchronousQueue<String>(); sQueue.add("aaa");

如果程序这样写,是肯定会报错的,因为SynchronousQueue里面是不能放元素的:

报的错: Exception in thread "main" java.lang.IllegalStateException: Queue full at java.util.AbstractQueue.add(AbstractQueue.java:98) at com.threadbasic015.SynchronousQueueDemo.main(SynchronousQueueDemo.java:8)

但是并不是说SynchronousQueue不能使用add()方法;

final SynchronousQueue<String> q = new SynchronousQueue<String>(); Thread t1 = new Thread(new Runnable() { @Override public void run() { try { System.out.println(q.take()); } catch (InterruptedException e) { e.printStackTrace(); } } }); t1.start(); Thread t2 = new Thread(new Runnable() { @Override public void run() { q.add("aaa"); } }); t2.start(); 这端代码是没有问题的,一个线程从SyhchronsQueue拿元素,一个线程往SynchronousQueue队列里面放元素,但是需要注意的是,add()方法,放的元素并不是 放在这个SynchronousQueue队列里面,而是直接拿给了take()方法去获取,如果没有take()方法,那么add()方法肯定是有问题的.

PriorityBlockingQueue:

基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定,也就是说传入队列的对象必须实现Comparable接口),在实现 PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁,他是一个无界的队列 这个队列是的元素是有优先级的,队列的元素要实现Comparaable接口,自定义比较的方法, 例子: public class Task implements Comparable<Task> { private int id; private String 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 int compareTo(Task task) { System.out.println("this:" + this); System.out.println("task:" + task); System.out.println("==============="); return this.id > task.id ? 1 : (this.id < task.id ? -1 : 0); } @Override public String toString() { return "Task [id=" + id + ", name=" + name + "]"; } }

---

public class UsePriorityBlockingQueue { public static void main(String[] args) throws InterruptedException { PriorityBlockingQueue<Task> q = new PriorityBlockingQueue<>(); Task t1 = new Task(); t1.setName("第一个元素"); t1.setId(3); /// Task t2 = new Task(); t2.setName("第二个元素"); t2.setId(2); /// Task t3 = new Task(); t3.setName("第三个元素"); t3.setId(1); // Task t4 = new Task(); t4.setName("第四个元素"); t4.setId(5); // Task t5 = new Task(); t5.setName("第五个元素"); t5.setId(4); q.add(t1); q.add(t2); q.add(t3); q.add(t4); q.add(t5); System.out.println(q); Task take = q.take(); System.out.println(take); System.out.println(q); } }

take方法回去找优先级中最高的元素

> 元素放进队列,开始是没有顺序的,当调用队列的take()方法,取数据,那么队列里面的元素就进行了一次比较,注意里面没有进行排序,一调用take()方法,把优 先级最高的元素取出来

DelayQueue:

带有延迟时间的Queue,其中的元素只有当其指定的延迟时间到了,才能够从队列中获取元素.DelayQueue中元素必须实现Delayed接口,DelayQueue是一个 没有大小限制的队列,应用场景很多,比如对缓存超时的数据进行移除,任务超时处理,空闲连接的关闭等待.

现在模拟一个例子:

网吧上网的例子,假设网吧开始营业,路人甲来上网交了1块钱,上了1小时,过了一段时间路人乙也来上网了,交了10块钱,上了10小时,过了一段时间路人丙来了, 交了5块钱,上了5小时.上网的时间跟交的钱成正比,这个事件相当于是队列中的延迟时间

实现:定义一个网名类(WangMing)实现了Delayed接口这个类要放在DelayQueue队列里面:

public class WangMing implements Delayed { private String name; private String id; //身份证号 private Long endTime; // 截止时间 private TimeUnit timeUnit = TimeUnit.SECONDS; public WangMing(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 long getDelay(TimeUnit unit) { System.out.println(this.getName() + " -当前时间:" + System.currentTimeMillis()); return endTime - System.currentTimeMillis(); } /** *相互比较,进行排序 */ @Override public int compareTo(Delayed delayed) { WangMing wangMing = (WangMing) delayed; return this.getDelay(timeUnit) - delayed.getDelay(timeUnit) > 0 ? 1:0; } }

定义网吧(WangBa)类:

public class WangBa implements Runnable { private DelayQueue<WangMing> queue = new DelayQueue<WangMing>(); private boolean yingye = true; public void shangji(String name, String id, int money) { WangMing man = new WangMing(name, id, 1000*money + System.currentTimeMillis()); System.out.println("网民:" + name +"开始上机了 - " + man.getEndTime() +"后下机"); this.queue.add(man); } public void xiaxian(WangMing man) { System.out.println(man.getName() + "下线了"); } @Override public void run() { while(yingye) { try { WangMing wangMing = queue.take(); xiaxian(wangMing); } catch (Exception e) { e.printStackTrace(); } } } public static void main(String[] args) { try { WangBa wangBa = new WangBa(); System.out.println("网吧开始营业了"); Thread shangwang = new Thread(wangBa); shangwang.start(); wangBa.shangji("路人甲", "123", 2); wangBa.shangji("路人乙", "234", 5); wangBa.shangji("路人丙", "345", 3); } catch (Exception e) { e.printStackTrace(); } } }

分析:

网吧类实现了Runnable接口,它的run方法,是要去DelaeyQuee队列中拿元素,如果,能够拿到元素,说明里面有元素的延迟时间已经用完了,就要下机.如果拿不到 元素,就一直在等着(这是一个死循环), WangMing实现了Delayed接口:有连个必须要实现的方法 1.getDelay()这个方法,只要这个元素没有超过延迟时间,就要一直运行这个方法,因为要通过这个方法来不断进行判断,是否过了延迟时间 2.comparedTo()这个方法,是DelayQueue内部元素需要实现排序要用的比较的方法

> 打印结果: 网吧开始营业了 网民:路人甲开始上机了 - 1524627051663后下机 网民:路人乙开始上机了 - 1524627058664后下机 网民:路人丙开始上机了 - 1524627056664后下机 路人甲下线了 路人丙下线了 路人乙下线了