多线程与并发----阻塞队列的应用

电玩女神 2021-09-15 10:20 330阅读 0赞

**一、队列**

**    1、队列分为固定长度的队列和不固定长度的队列；**

**    2、固定长度的队列，若放满了还要放，阻塞式队列就会等待，直到有数据取出，空出位置后才继续放；**

**    3、固定长度的队列，若放满了还要放，非阻塞式队列不能等待就只能报错了。**

**二、阻塞队列（BlockingQueue）**

> > >     public interface BlockingQueue<E>
> > > 
> > > extends [Queue][]<E>

支持两个附加操作的 [`Queue`][Queue]，这两个操作是：获取元素时等待队列变为非空，以及存储元素时等待空间变得可用。

BlockingQueue 方法以四种形式出现，对于不能立即满足但可能在将来某一时刻可以满足的操作，这四种形式的处理方式不同：第一种是抛出一个异常，第二种是返回一个特殊值（null 或 false，具体取决于操作），第三种是在操作可以成功前，无限期地阻塞当前线程，第四种是在放弃前只在给定的最大时间限制内阻塞。下表中总结了这些方法：

> > >   
> > > 
> > > 
> > > **![20180513201452261][]**

**注意：**

1、BlockingQueue 不接受 null 元素。试图 add、put 或 offer 一个 null 元素时，某些实现会抛出 NullPointerException。null 被用作指示 poll 操作失败的警戒值。

2、BlockingQueue 可以是限定容量的。它在任意给定时间都可以有一个 remainingCapacity，超出此容量，便无法无阻塞地 put 附加元素。没有任何内部容量约束的 BlockingQueue 总是报告 Integer.MAX\_VALUE 的剩余容量。

3、BlockingQueue实现主要用于生产者-使用者队列，但它另外还支持 [`Collection`][Collection] 接口。因此，举例来说，使用 remove(x) 从队列中移除任意一个元素是有可能的。然而，这种操作通常不会有效执行，只能有计划地偶尔使用，比如在取消排队信息时。

4、BlockingQueue 实现是线程安全的。所有排队方法都可以使用内部锁或其他形式的并发控制来自动达到它们的目的。然而，大量的 Collection 操作（addAll、containsAll、retainAll 和 removeAll）没有必要自动执行，除非在实现中特别说明。因此，举例来说，在只添加了 c 中的一些元素后，addAll(c) 有可能失败（抛出一个异常）。

5、BlockingQueue 实质上不 支持使用任何一种“close”或“shutdown”操作来指示不再添加任何项。这种功能的需求和使用有依赖于实现的倾向。例如，一种常用的策略是：对于生产者，插入特殊的 end-of-stream 或 poison 对象，并根据使用者获取这些对象的时间来对它们进行解释。

**三、实例**

**实例1：用长度为4的队列演示阻塞队列的功能和效果**

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class BlockingQueueTest {
    
    	public static void main(String[] args) {
    		final BlockingQueue queue = new ArrayBlockingQueue(4);
    		for(int i=0;i<2;i++){
    			new Thread(){
    				public void run(){
    					while(true){
    						try {
    							Thread.sleep((long)(Math.random()*1000));
    							System.out.println(Thread.currentThread().getName() + "准备放数据!");							
    							queue.put(1);
    							System.out.println(Thread.currentThread().getName() + "已经放了数据，" + 							
    										"队列目前有" + queue.size() + "个数据");
    						} catch (InterruptedException e) {
    							e.printStackTrace();
    						}
    
    					}
    				}
    				
    			}.start();
    		}
    		
    		new Thread(){
    			public void run(){
    				while(true){
    					try {
    						//将此处的睡眠时间分别改为100和1000，观察运行结果
    						Thread.sleep(1000);
    						System.out.println(Thread.currentThread().getName() + "准备取数据!");
    						queue.take();
    						System.out.println(Thread.currentThread().getName() + "已经取走数据，" + 							
    								"队列目前有" + queue.size() + "个数据");					
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    			}
    			
    		}.start();			
    	}
    }

运行部分结果为：

> Thread-1准备放数据!  
> Thread-1已经放了数据，队列目前有1个数据  
> Thread-0准备放数据!  
> Thread-0已经放了数据，队列目前有2个数据  
> Thread-1准备放数据!  
> Thread-1已经放了数据，队列目前有3个数据  
> Thread-0准备放数据!  
> Thread-0已经放了数据，队列目前有4个数据  
> Thread-2准备取数据!  
> Thread-2已经取走数据，队列目前有3个数据  
>

** 实例2：用两个具有 1 个空间的队列来实现同步通知的功能（也即是主线程运行一会，子线程运行一会，模拟一个队列已满，另一个队列是空的）**

>     import java.util.concurrent.ArrayBlockingQueue;
>     import java.util.concurrent.BlockingQueue;
>     
>     public class BlockingQueueCommunication {
>     
>     	public static void main(String[] args) {
>     		
>     		final Business business=new Business();
>     		
>     		new Thread(new Runnable() {
>     			
>     			public void run() {
>     				for(int i=0;i<5;i++){	
>     						business.sub(i);
>     				}
>     				
>     			}
>     			
>     		}).start();
>     		
>     		for(int i=0;i<5;i++){
>     			business.main(i);
>     		}
>     	}
>     	
>     	static class Business{
>     		//构造两个阻塞队列，长度均为1
>     		BlockingQueue<Integer> queue1=new ArrayBlockingQueue<Integer>(1);
>     		BlockingQueue<Integer> queue2=new ArrayBlockingQueue<Integer>(1);
>     		
>     		//用构造代码块给阻塞队列2赋初值，赋值后阻塞队列2已满
>     		{
>     			try {
>     				queue2.put(1);
>     			} catch (InterruptedException e) {
>     				e.printStackTrace();
>     			}
>     		}
>     		
>     		public void sub(int i){
>     			try {
>     				queue1.put(1);//向阻塞队列1中添加元素
>     			} catch (InterruptedException e) {
>     				e.printStackTrace();
>     			}
>     			for(int j=0;j<10;j++){
>     				System.out.println("sub thread sequence of"+j+",loop of"+i);
>     			}
>     			
>     			try {
>     				queue2.take();//从阻塞队列2中取出元素
>     			} catch (InterruptedException e) {
>     				e.printStackTrace();
>     			}
>     		}
>     		
>     		public void main(int i){
>     			try {
>     				queue2.put(1);//向阻塞队列2中添加元素
>     			} catch (InterruptedException e) {
>     				e.printStackTrace();
>     			}
>     			for(int j=0;j<10;j++){
>     				System.out.println("main thread sequence of"+j+",loop of"+i);
>     			}
>     			
>     			try {
>     				queue1.take();//从阻塞队列1中取出元素
>     			} catch (InterruptedException e) {
>     				e.printStackTrace();
>     			}
>     		}
>     	}
>     }

> 运行部分结果为：

> > sub thread sequence of0,loop of0  
> > sub thread sequence of1,loop of0  
> > sub thread sequence of2,loop of0  
> > sub thread sequence of3,loop of0  
> > sub thread sequence of4,loop of0  
> > sub thread sequence of5,loop of0  
> > sub thread sequence of6,loop of0  
> > sub thread sequence of7,loop of0  
> > sub thread sequence of8,loop of0  
> > sub thread sequence of9,loop of0  
> > main thread sequence of0,loop of0  
> > main thread sequence of1,loop of0  
> > main thread sequence of2,loop of0  
> > main thread sequence of3,loop of0  
> > main thread sequence of4,loop of0  
> > main thread sequence of5,loop of0  
> > main thread sequence of6,loop of0  
> > main thread sequence of7,loop of0  
> > main thread sequence of8,loop of0  
> > main thread sequence of9,loop of0  
> >

[Queue]: http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/Queue.html
[20180513201452261]: https://img-blog.csdn.net/20180513201452261
[Collection]: http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/Collection.html