Java并发编程---并发类容器(Queue容器)

﹏ヽ暗。殇╰゛Y 2022-06-02 03:06 480阅读 0赞

**一.概念**

**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();
    		}
    	}
    
    }

[Center]: /images/20220602/c63116ee75e14809a32e138dab28a600.png