并发 Queue，ConcurrentLinkedQueue，BlockingQueue，PriorityBlockingQueue，DelayQueue_综合

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

ConcurrentLinkedQueue

这是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常 ConcurrentLinkedQueue 性能好于 BlockingQueue。它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许 null 元素
ConcurrentLinkedQueue 重要方法： add()和 offer()都是加入元素的方法（在 ConcurrentLinkedQueue 中，这俩个方法没有任何区别） poll()和 peek()都是取头元素节点，区别在于前者会删除元素，后者不会。

// 高性能无阻塞无界队列：ConcurrentLinkedQueue 
ConcurrentLinkedQueue<String> clq = new ConcurrentLinkedQueue<>(); clq.offer("a"); 
clq.offer("b"); 
clq.offer("c"); 
clq.offer("d"); 
clq.add("e"); 
System.out.println(clq.poll()); // a 从头部取出元素，并从队列里删除 
System.out.println(clq.size()); // 4 
System.out.println(clq.peek()); // b 
System.out.println(clq.size()); // 4

BlockingQueue 接口

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

ArrayBlockingQueue<String> array = new ArrayBlockingQueue<>(5);array.put("a");array.put("b");array.add("c");array.add("d");array.add("e");// array.add("f");System.out.println(array.offer("a", 3, TimeUnit.SECONDS));

LinkedBlockingQueue：阻塞无界队列，可声明长度
LinkedBlockingQueue：基于链表的阻塞队列，同 ArrayBlockingQueue 类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），LinkedBlockingQueue 之所以能够高效的处理并发数据，是因为其内部实现采用分离锁（读写分离两个锁），从而实现生产者和消费者操作的完全并行运行。他是一个无界队列。
SynchronousQueue：无缓冲队列
SynchronousQueue：一种没有缓冲的队列，生产者产生的数据直接会被消费者获取并消费。
不同的场景，使用不同的队列。
在这里插入图片描述
完整代码如下：


public class UseQueue {
    public static void main(String[] args) throws Exception {
    // 高性能无阻塞无界队列：ConcurrentLinkedQueueConcurrentLinkedQueue<String> clq = new ConcurrentLinkedQueue<>();clq.offer("a");clq.offer("b");clq.offer("c");clq.offer("d");clq.add("e");System.out.println(clq.size());    // 5System.out.println(clq.poll());    // a 从头部取出元素，并从队列里删除System.out.println(clq.size());    // 4System.out.println(clq.peek());    // bSystem.out.println(clq.size());    // 4System.out.println("------------------------------------------------");// 阻塞有界队列ArrayBlockingQueue<String> array = new ArrayBlockingQueue<>(5);array.put("a");array.put("b");array.add("c");array.add("d");array.add("e");// array.add("f");System.out.println(array.offer("a", 3, TimeUnit.SECONDS));System.out.println("------------------------------------------------");// 阻塞无界队列，可声明长度LinkedBlockingQueue<String> lbq = new LinkedBlockingQueue<>();lbq.offer("a");lbq.offer("b");lbq.offer("c");lbq.offer("d");lbq.offer("e");lbq.add("f");System.out.println(lbq.size());lbq.forEach(System.out::println);System.out.println("------------------------------------------------");List<String> list = new ArrayList<>();System.out.println(lbq.drainTo(list, 3));System.out.println(list.size());list.forEach(System.out::println);System.out.println("------------------------------------------------");// 无缓冲队列final SynchronousQueue<String> q = new SynchronousQueue<>();Thread t1 = new Thread(() -> {
    try {
    System.out.println(q.take());} catch (InterruptedException e) {
    e.printStackTrace();}});t1.start();Thread t2 = new Thread(() -> q.add("asdasd"));t2.start();}
}

PriorityBlockingQueue：基于优先级的阻塞队列
PriorityBlockingQueue：基于优先级的阻塞队列（优先级的判断通过构造函数传入的 Compatory 对象来决定，也就是说传入队列的对象必须实现 Comparable 接口，在实现 PriorityBlockingQueue 时，内部控制线程同步的锁采用的是公平锁，他也是一个无界的队列。
任务接口代码：


public class Task implements Comparable<Task> {
    private int id;private String name;public Task() {
    }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;}@Overridepublic int compareTo(Task task) {
    return this.id > task.id ? 1 : (this.id < task.id ? -1 : 0);}@Overridepublic 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(3, "t1");Task t2 = new Task(5, "t2");Task t3 = new Task(2, "t3");Task t4 = new Task(8, "t4");Task t5 = new Task(6, "t5");q.put(t1);q.put(t2);q.add(t3);q.add(t4);q.add(t5);q.forEach(System.out::println);System.out.println("------------------------");System.out.println(q.take());System.out.println(q.take());System.out.println(q.take());System.out.println(q.take());System.out.println(q.take());System.out.println(q);}
}

DelayQueue：带有延迟时间的 Queue
DelayQueue：带有延迟时间的 Queue，其中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。 DelayQueue 中的元素必须实现Delayed 接口， Delay Queue 是一个没有大小限制的队列，应用场景很多，比如对缓存超时的数据进行移除、任务超时处理、空闲连接的关闭等等。
该队列使用一个网吧上网的例子来展示：
WangBa.java

public class WangBa implements Runnable {
    private DelayQueue<Wangmin> queue = new DelayQueue<>();public boolean yinye = true;// 是否营业public void shangji(String name, String id, int money) {
    Wangmin man = new Wangmin(name, id, 1000 * money + System.currentTimeMillis());System.out.println("网名" + man.getName() + " 身份证" + man.getId() + "交钱" + money + "块,开始上机...");this.queue.add(man);}public void xiaji(Wangmin man) {
    System.out.println("网名" + man.getName() + " 身份证" + man.getId() + "时间到下机...");}@Overridepublic void run() {
    while (yinye) {
    try {
    Wangmin man = queue.take();xiaji(man);} catch (InterruptedException e) {
    e.printStackTrace();}}}public static void main(String args[]) {
    try {
    System.out.println("网吧开始营业");WangBa wangBa = new WangBa();Thread shangwang = new Thread(wangBa);shangwang.start();wangBa.shangji("路人甲", "123", 1);wangBa.shangji("路人乙", "234", 10);wangBa.shangji("路人丙", "345", 5);} catch (Exception e) {
    e.printStackTrace();}}
}

Wangmin.java


public class Wangmin implements Delayed {
    // 身份证private String id;// 姓名private String name;// 下机时间private long endTime;// 定义时间类private TimeUnit timeUnit = TimeUnit.SECONDS;public Wangmin(String id, String name, long endTime) {
    this.id = id;this.name = name;this.endTime = endTime;}public String getName() {
    return this.name;}public String getId() {
    return this.id;}/*** 用来判断是否到了截止时间*/@Overridepublic long getDelay(TimeUnit unit) {
    return endTime - System.currentTimeMillis();}/*** 相互比较排序用*/@Overridepublic int compareTo(Delayed delayed) {
    Wangmin w = (Wangmin) delayed;return this.getDelay(this.timeUnit) - w.getDelay(this.timeUnit) > 0 ? 1 : 0;}}

我们使用实现Runnable 接口来实现我们的业务。
在这里插入图片描述