JAVA并发工具包-阻塞队列

￡神魔★判官ぃ 2022-03-20 08:07 272阅读 0赞

java.util.concurrent - Java 并发工具包  
Java 5 新添加了并发访问包java.util.concurrent 包。包含有一系列能够让 Java 的并发编程变得更加简单轻松的类，包括阻塞对象、锁、可用重入读写锁、线程池、写时复制集合等。

## 阻塞对象BlockingQueue ##

![在这里插入图片描述][20190211211141255.png]  
BlockingQueue是阻塞队列的接口，它可以保证多线程同时访问此对象时数据保持一致性，有如下特性：多个线程共享此对象时，向队列里存放元素的线程，在队列已满后自动阻塞；从队列获取元素的线程，在队列已空时自动阻塞。  
BlockingQueue接口提供的方法有：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70]

四类方法不同特性：  
抛异常：如果方法无法立即执行，抛一个异常。  
特定值：如果方法无法立即执行，返回一个特定的值(true / false)。  
阻塞：如果方法无法立即执行，该方法调用线程发生阻塞，直到条件允许后方可执行。  
超时：如果方法无法立即执行，该方法调用线程发生阻塞，直到条件允许后方可执行，但等待时间不会超过给定值。返回一个特定值 (true / false)。

BlockingQueue接口实现类：  
ArrayBlockingQueue  
DelayQueue  
LinkedBlockingQueue  
PriorityBlockingQueue  
SynchronousQueue

## 例程　BlockingQueueTest ##

package com.hk;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class BlockingQueueTest {
    	public static void main(String[] args)
    	{
    		try
    		{
    	        BlockingQueue queue = new ArrayBlockingQueue(1024);
    	        
    	        ProducerTh producer = new ProducerTh(queue);
    	        ConsumerTh consumer = new ConsumerTh(queue);
    	        new Thread(producer).start();
    	        new Thread(consumer).start();
    	 
    	        Thread.sleep(1000);
    		}
    		catch(Exception e)
    		{
    			e.printStackTrace();
    		}
    	}
    }
    ProducerTh
    package com.hk;
    
    import java.util.concurrent.BlockingQueue;
    
    public class ProducerTh implements Runnable{
    	 
        protected BlockingQueue queue = null;
     
        public ProducerTh(BlockingQueue queue) {
            this.queue = queue;
        }
     
        public void run() {
            try {
                queue.put("1");
                Thread.sleep(1000);
                queue.put("2");
                Thread.sleep(1000);
                queue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    ConsumerTh
    package com.hk;
    
    import java.util.concurrent.BlockingQueue;
    
    public class ConsumerTh implements Runnable{
    	 
        protected BlockingQueue queue = null;
     
        public ConsumerTh(BlockingQueue queue) {
            this.queue = queue;
        }
     
        public void run() {
            try {
                System.out.println(queue.take());
                System.out.println(queue.take());
                System.out.println(queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

## ArrayBlockingQueue ##

ArrayBlockingQueue 类实现BlockingQueue 接口,内部使用数组存储对象，数组一旦初始化不能被修改，ArrayBlockingQueue 内部FIFO(先进先出)的顺序对元素进行存储。

## DelayQueue ##

DelayQueue 实现BlockingQueue 接口，DelayQueue对元素保留一段特定的延迟到期后可以获取，其元素必须实现接口java.util.concurrent.Delayed，

package com.hk;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    public class DelayQueueTest {
    	public static void main(String[] args)
    	{
    		try
    		{
    			DelayQueue<MyEle> dqueue = new DelayQueue<MyEle>();
    			dqueue.put(new MyEle(600L));
    			dqueue.put(new MyEle(200L));
    			MyEle ele1 = dqueue.take();
    			System.out.println("ele1="+ele1);
    			
    			MyEle ele2 = dqueue.take();
    			
    			System.out.println("ele2="+ele2);
    		}
    		catch(Exception e)
    		{
    			e.printStackTrace();
    		}
    	}
    }
    
    class MyEle implements Delayed
    {
    	//对象创建时间
    	private long createTimeNano = System.nanoTime();
    	//对象持续纳秒数
    	private long count = 0;
    	
    	public MyEle(long count)
    	{
    		this.count = count;
    	}
    	
    	public long getCount()
    	{
    		return count;
    	}
    
    	public int compareTo(Delayed o) {
    		if(! (o instanceof MyEle))
    			return -1;
    		MyEle ele = (MyEle)o;
    		if(count == ele.count)
    			return 0;
    		if(count >= ele.count)
    			return 1;
    		else
    			return -1;
    	}
    
    	public long getDelay(TimeUnit unit) {
    		//计算对象自创建起经过多少时间(纳秒)
    		long l = createTimeNano + unit.convert(count,  TimeUnit.NANOSECONDS) - System.nanoTime();
    		return l;
    	}
    	
    	public String toString()
    	{
    		return "MyEle[count="+count+"]";
    	}
    }

![在这里插入图片描述][20190211211509948.png]

## LinkedBlockingQueue ##

LinkedBlockingQueue 类实现BlockingQueue 接口。LinkedBlockingQueue 内部以一个链式结构(链接节点)对其元素进行存储。如果需要的话，这一链式结构可以选择一个上限。如果没有定义上限，将使用 Integer.MAX\_VALUE 作为上限。  
LinkedBlockingQueue 内部以 FIFO(先进先出)的顺序对元素进行存储。

package com.hk;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class BlockingQueueTest {
    	public static void main(String[] args)
    	{
    		try
    		{
    			BlockingQueue<String> queue1 = new LinkedBlockingQueue<String>();
    			BlockingQueue<String> queue2   = new LinkedBlockingQueue<String>(1024);
    			
    			for(int i=0;i<2000;i++)
    				queue1.put("Str"+i);
    			System.out.println("Queue1 exit...");
    			for(int i=0;i<2000;i++)
    			{
    				queue2.put("Str"+i);
    				System.out.println("i="+i);
    			}
    			System.out.println("Queue2 exit...");
    		}
    		catch(Exception e)
    		{
    			e.printStackTrace();
    		}
    	}
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 1]  
第二个队列长度为1024，当添加到第1025个元素是线程阻塞。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 2]

## 优先级阻塞队列 PriorityBlockingQueue ##

PriorityBlockingQueue 类实现BlockingQueue接口。存放的元素必须实现 java.lang.Comparable接口,队列中元素的排列顺序取决于Comparable实现

package com.hk;
    
    import java.util.Iterator;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.PriorityBlockingQueue;
    
    public class BlockingQueueTest {
    	public static void main(String[] args)
    	{
    		try
    		{
    			PriorityBlockingQueue<MyEle2> queue1 = new PriorityBlockingQueue<MyEle2>(100);
    			
    			for(int i=0;i<100;i++)
    				queue1.put(new MyEle2(i));
    			for(int i=0;i<100;i++)
    			{
    				MyEle2 ele2 = queue1.take();
    				System.out.println(ele2);
    			}
    		}
    		catch(Exception e)
    		{
    			e.printStackTrace();
    		}
    	}
    }
    class MyEle2 implements Comparable
    {
    	//排序号
    	private long count = 0;
    	
    	public MyEle2(long count)
    	{
    		this.count = count;
    	}
    	
    	public long getCount()
    	{
    		return count;
    	}
    
    	public int compareTo(Object o) {
    		if(! (o instanceof MyEle2))
    			return -1;
    		MyEle2 ele = (MyEle2)o;
    		if(count == ele.count)
    			return 0;
    		if(count >= ele.count)
    			return 1;
    		else
    			return -1;
    	}
    	
    	public String toString()
    	{
    		return "MyEle2[count="+count+"]";
    	}
    }

## 同步队列 SynchronousQueue ##

SynchronousQueue 类实现BlockingQueue 接口。  
SynchronousQueue 是一个特殊的队列，它的内部同时只能够容纳单个元素。如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。

## 阻塞双端队列 BlockingDeque ##

java.util.concurrent 包里的 BlockingDeque 接口表示一个线程安放入和提取实例的双端队列，BlockingDeque 类是一个双端队列，可以从队列头部和尾部存放和提取元素。  
![在这里插入图片描述][20190211211803227.png]  
BlockingDeque提供从头部或尾部存取数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 3]  
四类方法不同特性：  
抛异常：如果方法无法立即执行，抛一个异常。  
特定值：如果方法无法立即执行，返回一个特定的值(true / false)。  
阻塞：如果方法无法立即执行，该方法调用线程发生阻塞，直到条件允许后方可执行。  
超时：如果方法无法立即执行，该方法调用线程发生阻塞，直到条件允许后方可执行，但等待时间不会超过给定值。返回一个特定值 (true / false)。

## ArrayBockingQueue与LinkedBockingQueue区别 ##

1.队列大小不同，ArrayBlockingQueue使用数组存储，必须在初始化时设定数组长度，而LinkedBlockingQueue使用链表存储，可以是有界的，也可以是无界的(Integer.MAX\_VALUE)，如果LinkedBockingQueue不设大小，当添加速度大于移除速度时，在无界的情况下，可能会造成内存溢出等问题。

2.数据存储方式不同，ArrayBlockingQueue采用的是数组作为数据存储容器，而LinkedBlockingQueue采用的则是以Node节点作为连接对象的链表。

3.ArrayBlockingQueue采用的是数组的存储容器，在插入或删除元素时不会产生或销毁任何额外的对象实例，LinkedBlockingQueue生成一个额外的Node对象。大批量操作数据时，对于GC可能存在较大影响。

4.两者的实现队列添加或移除的锁不一样，ArrayBlockingQueue实现的队列中的锁是没有分离的，即添加操作和移除操作采用的同一个ReenterLock锁，而LinkedBlockingQueue实现的队列中的锁是分离的，其添加采用的是putLock，移除采用的则是takeLock，这样能大大提高队列的吞吐量，也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

[20190211211141255.png]: /images/20220320/f4020a587392441591e60a43f6bbf632.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70]: /images/20220320/a9a21365b99e4edb80ab21ed25090f88.png
[20190211211509948.png]: /images/20220320/bf9c950a412343e0ba73a461ac7ffe16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 1]: /images/20220320/b384aa474bd54499869cc369551d6295.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 2]: /images/20220320/d9a339a3e9ab48feb1bd902bdc5d01c3.png
[20190211211803227.png]: /images/20220320/7ad86c4365474a3b853839b91f6be630.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FpeGlhbmdfY2hlbg_size_16_color_FFFFFF_t_70 3]: /images/20220320/03e21e37efbc4291bb0dc6937870d149.png