手撕阻塞队列BlockingQueue

绝地灬酷狼 2023-01-19 09:38 6阅读 0赞

# 一、什么是阻塞队列 #

**BlockingQueue 继承了 Queue 接口，是队列的一种**。  
当阻塞队列为空时，从队列中获取元素的操作将会被阻塞；当阻塞队列满了，往队列添加元素的操作将会被阻塞。  
在多线程中，阻塞的意思是，在某些情况下会挂起线程，一旦条件成熟，被阻塞的线程就会被自动唤醒。

# 二、BlockQueue常用的方法 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70]  
根据插入和取出两种类型的操作，具体分为下面几些类型：  
![在这里插入图片描述][20210509151615443.png]  
**抛出异常**：这时候插入和取出在不能立即被执行的时候就会抛出异常。  
**特殊值**：插入和取出在不能被立即执行的情况下会返回一个特殊的值（true 或者 false）  
**阻塞**：插入和取出操作在不能被立即执行时会阻塞线程，直到条件成熟，被其他线程唤醒  
**超时**：插入和取出操作在不能立即执行的时候会被阻塞一定的时候，如果在指定的时间内没有被执行，那么会返回一个特殊值。  
单单从操作的维度来看的话，还是会有点错，因为有些方法是阻塞方法，有些方法不是，我们从阻塞和不阻塞的维度再来一次划分：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 1]

# 三、常见的阻塞队列 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 2]  
**ArrayBlockingQueue**和**LinkedBlockingQueue**是最为常用的阻塞队列，前者使用一个有边界的数组来作为存储介质，而后者使用了一个没有边界的链表来存储数据。

PriorityBlockingQueue是一个优先阻塞队列。所谓优先队列，就是每次从队队列里面获取到的都是队列中优先级最高的，对于优先级，PriorityBlockingQueue需要你为插入其中的元素类型提供一个Comparator，PriorityBlockingQueue使用这个Comparator来确定元素之间的优先级关系。底层的数据结构是堆，也就是我们数据结构中的那个堆。

DelayQueue是一个延时队列，所谓延时队列就是消费线程将会延时一段时间来消费元素。

**SynchronousQueue**是最为复杂的阻塞队列。SynchronousQueue和前面分析的阻塞队列都不同，因为SynchronousQueue不存在容量的说法，任何插入操作都需要等待其他线程来消费，否则就会阻塞等待。

# 四、案例说明 #

SynchronousQueue实现生产者消费者功能，参考: [生产者消费者四种方式实现][Link 1].

package com.zhang;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.TimeUnit;
    
    public class BlockingQueueTest { 
    
        public static void main(String[] args) { 
            //SynchronousQueue不存在容量的说法，即只存一个元素,任何插入操作都需要等待其他线程来消费，否则就会阻塞等待
            BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
            new Thread(()->{ 
                try { 
                    System.out.println(Thread.currentThread().getName() + "\t ---> 生产 put 1");
                    blockingQueue.put("1");
                    System.out.println(Thread.currentThread().getName() + "\t ---> 生产 put 2");
                    blockingQueue.put("2");
                    System.out.println(Thread.currentThread().getName() + "\t ---> 生产 put 3");
                    blockingQueue.put("3");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            },"生产者").start();
    
            new Thread(()->{ 
                try { 
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName() + "\t ---> 消费"+blockingQueue.take());
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName() + "\t ---> 消费"+blockingQueue.take());
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName() + "\t ---> 消费"+blockingQueue.take());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            },"消费者").start();
        }
    }

生产者消费者模型结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 3]  
LinkedBlockingQueue在线程池方面的应用，参考[Java 线程池][Java].

package com.test;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    import org.apache.tomcat.util.collections.SynchronizedQueue;
    
    public class ThreadPoolLinkedBlockingQueueTest { 
    
    	 public static void main(String[] args) throws Exception {    
    	        
    		    Runnable runnable = new Runnable() { 
    				@Override
    				public void run() { 
    						try { 
    							Thread.sleep(2000);
    						} catch (InterruptedException e) { 
    							// TODO Auto-generated catch block
    							e.printStackTrace();
    						}
    						System.out.println(Thread.currentThread().getName()+" run ");
    				}
    			};
    			/* LinkedBlockingQueue当线程不超过核心线程数之时，会复用核心线程数（可以看出线程名称相同！） */
    	    	ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 4, 5, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>());  
    	    	executor.execute(runnable);
    	    	executor.execute(runnable);
    	    	executor.execute(runnable);
    	    	System.out.println("先开3个线程-----------");
    	    	System.out.println("线程池核心线程数："+executor.getCorePoolSize());
    	    	System.out.println("线程池中线程数："+executor.getPoolSize());
    	    	System.out.println("线程池中队列任务数："+executor.getQueue().size());
    	    	executor.execute(runnable);
    	    	executor.execute(runnable);
    	    	executor.execute(runnable);
    	    	System.out.println("再开3个线程-----------");
    	    	System.out.println("线程池核心线程数："+executor.getCorePoolSize());
    	    	System.out.println("线程池中线程数："+executor.getPoolSize());
    	    	System.out.println("线程池中队列任务数："+executor.getQueue().size());
    	    	executor.shutdown();
    	    }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 4]  
LinkedBlockingQueue不受最大线程数影响，但是当其queue有大小限制则会受影响！  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 5]  
**参考文章**  
https://baijiahao.baidu.com/s?id=1659873724613844132&wfr=spider&for=pc  
https://www.cnblogs.com/tjudzj/p/4454490.html  
https://blog.csdn.net/xiewenfeng520/article/details/107100303/

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70]: /images/20221021/e343ce9bc49d42d88c5e7c8b9415c100.png
[20210509151615443.png]: /images/20221021/d3da3480f5ea405f98f2ee3856eca887.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 1]: /images/20221021/a5a33b2d334146459969cc02fb074edd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 2]: /images/20221021/1a904d355ce347a783809b21485cb0cf.png
[Link 1]: https://blog.csdn.net/jike11231/article/details/112792709
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 3]: /images/20221021/96bba4437a25441ebede31f29e7fa191.png
[Java]: https://blog.csdn.net/jike11231/article/details/106971203
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 4]: /images/20221021/02ca8ba9cd9b48ffb93c9b2a841b4093.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70 5]: /images/20221021/7731c7d926824641a833d252fc09e2fe.png