Java并发编程---并发类容器(Queue容器)-蒲公英云

一.概念

1.1 并发Queue

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

Center

1.2 ConcurrentLinkedQueue(属于Queue队列接口下的方法)

ConcurrentLinkedQueue是一个适用于高并发场景的队列,通过无锁的方式,实现了高并发状态下的高性能,通常ConcurrentLinkedQueue性能好于BlockingQueue.它是一个基于链接节点的无界线程安全队列.该队列的元素遵循先进先出的原则.头是最先加入的,尾是最近加入的,该队列不允许null元素

ConcurrentLinkedQueue重要方法:

add()和offer()都是加入元素的方法(在ConcurrentLinkedQueue中,这两个方法没有任何区别)

poll()和peek()都是取头元素节点,区别在于前者会删除元素,后者不会

1.3 BlockingQueue(阻塞队列)接口

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

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

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

(4) PriorityBlockingQueue: 基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定,也就是说传入队列的对象必须实现Comparable接口),在实现PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁,他也是一个无界的队列.

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

二.代码示例

package com.thread.container;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
public class UseQueue {
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        //高性能无阻塞无界队列: ConcurrentLinkedQueue
        ConcurrentLinkedQueue<String> q = new ConcurrentLinkedQueue<String>();
        q.offer("a");
        q.offer("b");
        q.offer("c");
        q.offer("d");
        q.add("e");
        System.out.println(q.poll());                 //a    从头部取出元素,并从队列里删除
        System.out.println(q.size());                 //4
        System.out.println(q.peek());                 //b
        System.out.println(q.size());                 //4
        ArrayBlockingQueue<String> array = new ArrayBlockingQueue<String>(5);
        array.put("a");
        array.put("b");
        array.put("c");
        array.put("d");
        array.put("e");
        array.put("f");
        System.out.println(array.offer("a",3,TimeUnit.SECONDS));    //队列有界是5个,现在放入6个,所以会报Queue full(容器满了)的异常
        //阻塞队列
        LinkedBlockingQueue<String> q = new LinkedBlockingQueue<String>();
        q.offer("a");
        q.offer("b");
        q.offer("c");
        q.offer("d");
        q.offer("e");
        q.add("f");
        System.out.println(q.size());
        for(Iterator iterator = q.iterator(); iterator.hasNext();){
            String str = (String) iterator.next();
            System.out.println(str);
        }
        List<String> list = new ArrayList<String>();
        System.out.println(q.drainTo(list,3));
        System.out.println(list.size());
        for(String str : list){
            System.out.println(str);
        }
        //一种没有缓冲的队列,生产者产生的数据直接会被消费者获取并消费
        final SynchronousQueue<String> q = new SynchronousQueue<String>();
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                // TODO Auto-generated method stub
                try {
                    System.out.println("进入t1线程中,阻塞等待获取元素...");
                    System.out.println("消费: " + q.take());
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
        t1.start();
        TimeUnit.SECONDS.sleep(2);     //阻塞2秒钟
        Thread t2 = new Thread(new Runnable() {    
            @Override
            public void run() {
                // TODO Auto-generated method stub
                q.add("haige");
            }
        });
        t2.start();
    }
}

PriorityBlockingQueue容器的示例

package com.thread.container;
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;
    }
    public int compareTo(Task task) {
        // TODO Auto-generated method stub
        return this.id>task.id ? 1 : (this.id<task.id ? -1 : 0);
    }
    @Override
    public String toString() {
        return "Task [id=" + id + ", name=" + name + "]";
    }
}
package com.thread.container;
import java.util.concurrent.PriorityBlockingQueue;
/**
 * PriorityBlockingQueue示例
 * @author lhy
 * @time   2018.01.02
 */
public class UsePriorityBlockingQueue {
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
       PriorityBlockingQueue<Task> q = new PriorityBlockingQueue<Task>();
       Task t1 = new Task();
       t1.setId(3);
       t1.setName("haige");
       Task t2 = new Task();
       t2.setId(4);
       t2.setName("xiaozhang");
       Task t3 = new Task();
       t3.setId(1);
       t3.setName("xiaoke");
       q.add(t1);           //3
       q.add(t2);           //4
       q.add(t3);           //1
       //1  3  4
       System.out.println("容器: " + q);
       System.out.println(q.take().getId());
       System.out.println("容器: " + q);
       System.out.println(q.take().getId());
       System.out.println("容器: " + q);
       System.out.println(q.take().getId());
    }
}

DelayQueue示例

package com.thread.container;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class WangMin implements Delayed {
    private String name;
    //身份证
    private String id;
    //截止时间
    private long endTime;
    //定义时间工具类
    private TimeUnit timeUtil = TimeUnit.SECONDS;
    public WangMin(String name, String id, long endTime) {
        super();
        this.name = name;
        this.id = id;
        this.endTime = endTime;
    }
    public WangMin() {
        // TODO Auto-generated constructor stub
    }
    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 delayed) {
        // TODO Auto-generated method stub
        WangMin w = (WangMin) delayed;
        return this.getDelay(this.timeUtil) - w.getDelay(this.timeUtil) > 0 ?1:0 ;    
    }
    //用来判断是否到了截止时间(两者相减<=0,则说明到了下机时间)
    @Override
    public long getDelay(TimeUnit unit) {
        // TODO Auto-generated method stub
        return endTime - System.currentTimeMillis();
    }
}
package com.thread.container;
import java.util.concurrent.DelayQueue;
/**
 * DelayQueue示例
 * @author lhy
 * @ time 2018.01.02
 */
public class WangBa implements Runnable {
    private DelayQueue<WangMin> queue = new DelayQueue<WangMin>();
    public boolean yinyue = 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() + "时间到下机");
    }
    @Override
    public void run() {
        // TODO Auto-generated method stub
        while(yinyue){
            try {
                WangMin man = queue.take();
                Xiaji(man);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        try {
            System.out.println("网吧开始营业");
            WangBa haige = new WangBa();
            Thread shangwang = new Thread(haige);
            shangwang.start();
            haige.Shangji("yangxu", "123", 1);
            haige.Shangji("yangming", "123", 5);
            haige.Shangji("yangge", "123", 10);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}