充分了解java阻塞队列机制

1.阻塞队列

1.1 什么是 阻塞队列

public interface BlockingQueue<E> extends Queue<E> {
}

BlockingQueue继承了Queue的接口,是队列的一种,并且和Queue相比,BlockingQueue是线程安全的,多用于并发+并行编程,对于线程安全问题可以很好的解决.

下面是实现BlockingQueue接口的类

怕大家理解不方便,俺通过思维导图的方式给大家呈现

阻塞队列的典型例子就是BlockingQueue接口的实现类, 主要有六种实现
ArrayBlcokingQueue,LinkedBlockingQueue,SynchronousQueue,DelayQueue,PriorityBlockingQueue和 LinkedTransherQueue,它们各自有不同不同的特点。

1.2 阻塞队列的特点

在讲阻塞队列特点前,先给大家用图演示一下在没有阻塞队列时,服务器之间的联系.

1.服务器A将接受到的请求传输给服务器B,他们之间联系是单线联系,也就是服务器A可以直接访问到服务器B,这样做会有一个很大的缺点,我们假设服务器A崩溃了,那么由于服务器B是和服务器A是相关联的,所以服务器B也会收到一定量的影响,甚至是一起崩溃…

2.此时我们在来看这一张图,由于服务器A和B是密切关联的,所以当我想再让服务器A和C关联,我们不仅需要修改服务器A的代码,包括服务器B的代码我们也需要进行修改,此时如果再加上服务器D,E,F等等,经过这样的频繁修改代码,那便会对系统带来不可预估的损失.

3.所以我们在写代码时都会强调低耦合,给大家举例子来解释这个意思:

我们用苹果手机举例,由于苹果手机充电插口指定只有苹果官方的充电器才可以进行充电,所以我们可以看出,苹果手机如果想使用,只能依赖苹果官方充电器,如果没有这个充电器或者这个充电器坏了的话,那么苹果手机也就无法使用的.这就是高耦合,两者的依赖很深,谁都不能离开谁,其中一个坏掉,另一个也会收到影响.

我们再用安卓手机举例,由于安卓手机并没有指定必须是官方的充电器才可以充电,所以即使是这个充电器坏掉,俺也可以找到另一个充电器来平替,简单的叙述如下:若A与B存在依赖关系,那么当B发生改变时,A依然可以正常使用,此时就可以认为A与B时低耦合的.

那么我们如何解决这个耦合性高的问题呢?

俺们可以引入阻塞队列来降低它们之间的耦合性.

如下图:

1. 当我们引入阻塞队列后,就可以很优雅的解决耦合性高的问题.
   此时服务器A并不知道服务器B的存在,服务器A只认识阻塞队列,他的任务也就是将收到的请求添加到阻塞队列里面,服务器B同理,它也是只知道从阻塞队列里面读取请求,然后根据请求完成任务.此时不管是A,B那个服务器出现错误,另一个服务器也都不会收到影响.

2. 即使现在服务器C也从阻塞队列中读取请求,不过由于他们各个服务器之间并没有关联,所以服务器C的出现对其他服务器的影响也是微乎其微的.

3. 3.阻塞队列还有一个功能就是削峰填谷,什么意思呢?

我们假设服务器A平时收到的请求是1000条/s,但是突然今天收到的请求是平常的好多倍

• 当两个服务器没有使用阻塞队列时,服务器A的请求一股脑传给了服务器B,那么此时服务器B就会因为突然要处理的请求太多而导致程序崩溃.

  如图:不出意外,水杯里的水由于装不下就会溢出.
  

• 当服务器之间添加了阻塞队列作为中介时,虽然A突然增添了许多请求给到阻塞队列中,但是并不影响B读取请求的速率,就像是下图,

这是削峰添谷的曲线图,其中灰色部分就是将多余的黄色部分填充得来的.

小结:阻塞功能使得生产者和消费者两端的能力得以平衡，当有任何一端速度过快时，阻塞队列便会把过快的速度降下来。

1.3 阻塞队列常用方法

在阻塞队列中有很多的方法,而且非常相似,常用的8个方法主要以添加删除为主,主要分为三类:

1.抛出异常:add、remove、element
2.返回结果但是不抛出异常: offer、poll、peek
3.阻塞:take、put

1.3.1 抛出异常:add、remove、element

add方法是往队列里面添加一个元素，如果队列满了，就会抛出异常来提示我们队列已满。

//源码public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full");}

当插入元素失败时,就会抛出异常.测试代码如下:

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);System.out.println(blockingQueue.add(1));System.out.println(blockingQueue.add(1));System.out.println(blockingQueue.add(3));}
}

运行结果:

remove方法是删除元素，如果我们队列为空的时候又进行了删除操作，同样会报NoSuchElementException,且在删除操作成功后会返回被删除的值。

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);blockingQueue.add(1);blockingQueue.add(2);System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());blockingQueue.remove();}
}

这里我们指定容量为2，并且添加两个元素，然后删除三个元素。结果如下

运行结果:

element方法是返回队列的头节点，但是不会删除这个元素。当队列为空时同样会报NoSuchElementException的错误.

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);blockingQueue.element();}
}

此时我们对这个空队列返回队首元素.

运行结果:

1.3.2 返回结果但是不抛出异常offer、poll、peek

offer方法用来插入一个元素，如果插入成功会返回true，如果队列满了，再插入元素不会抛出异常但是会返回false。

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);System.out.println(blockingQueue.offer(1));System.out.println(blockingQueue.offer(1));System.out.println(blockingQueue.offer(1));}
}

此时队列的容量为2,当我们添加第三个元素之后就会返回false.

poll方法和remove方法是对应的都是删除元素，都会返回删除的元素，但是当队列为空时则会返回null.

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);System.out.println(blockingQueue.poll());}
}

此时队列里没有元素,我们再进行poll就会返回null.

peek方法和element方法对应，返回队列的头节点但并不删除，如果队列为空则直接返回null.

public class Test1 {public static void main(String[] args) {//创建一个只有两个容量的阻塞队列BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(2);System.out.println(blockingQueue.peek());}
}

此时队列里没有元素,如果进行peek会返回null.

带超时时间的offer和poll

offer(E e, long timeout, TimeUnit unit){
}

它有三个参数，分别是元素、超时时长和时间单位。通常情况下，这个方法会插入成功并且返回true；如果队列满了导致插入不成功，在调用带超时时间重载方法的offer的时候，则会等待指定的超时时间，如果到了时间依然没有插入成功，则返回false。

E poll(long timeout, TimeUnit unit){
}

这个带参数的poll和上面的offer类似。如果能够移除，便会立即返回这个节点的内容；如果超过了我们定义的超时时间依然没有元素可以移除，便会返回null作为提示。

1.3.3 阻塞put和take

put:添加一个元素,如果队列此时满了就会进行阻塞.
take:删除队首元素,如果队列为空就会阻塞

put方法的作用是插入元素，通常在队列没有满的时候是正常插入。如果队列满了无法继续插入，这时它不会立刻返回false和抛出异常，而是让插入的线程进入阻塞状态，直到队列里面有空闲空间了。此时队列就会让之前的线程解除阻塞状态，并把刚才那个元素添加进去。 take方法的作用是获取并移除队列的头节点。通常队列里面有元素会正常取出数据并移除；但是如果执行take的时候队列里无数据，则阻塞，直到队列里面有数据以后，就会立即解除阻塞状态，并且取到数据.

1.3.4 小结

ArrayBlockingQueue是一个基于数组实现的有界的阻塞队列。

几个要点

• ArrayBlockingQueue是一个用数组实现的队列，所以在效率上比链表结构的LinkedBlockingQueue要快一些，但是队列长度固定，不能扩展，入列和出列使用同一把锁。LinkedBlockingQueue是入列出列两把锁，读写分离。
• 先进先出，FIFO，队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素
• 新元素插入到队列的尾部，队列检索操作则是从队列头部开始获得元素
  利用重入锁来保证并发安全
• 初始化时必须传入容量，也就是数组的大小，不需要扩容，因为是初始化时指定容量，并循环利用数组，使用之前一定要慎重考虑好容量
• put(e)（put(e)时如果队列满了则使用notFull阻塞等待）、take()阻塞
• add(e)时如果队列满了则抛出异常
• remove()时如果队列为空则抛出异常
• offer(e)时如果队列满了则返回false
• poll()时如果队列为空则返回null
• poll(timeout, unit)时如果队列为空则阻塞等待一段时间后如果还为空就返回null
  只使用了一个锁来控制入队出队，效率较低

1.4 常见的阻塞队列

1.4.1 ArrayListBlockingQueue

常见的构造方法如下

下面是各个参数的意思.

public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c){
}

1.创建一个ArrayBlockingQueue，该队列具有给定（固定）容量、指定的访问策略，最初包含给定集合的元素，并按集合迭代器的遍历顺序添加。
2.参数：
capacity–此队列的容量
fair–如果为true，则在插入或删除时被阻止的线程的队列访问将按FIFO顺序进行处理(公平的,先来先处理)；如果为false，则未指定访问顺序(也就是非公平的,其他线程就有可以插队的可能)。
c–最初包含的元素集合

对于ArrayListBlockingQueue类,它的内部是通过一个循环队列来实现的,这也就导致了它无法扩容,所以我们在创建这个队列时,一定要慎重考虑好容量.

那么我们该如何自己实现一个阻塞队列呢?

1. 我们先实现一个普通的循环队列

“循环队列的优点:可以有效的利用资源。用数组实现队列时,如果不移动,随着数据的不断读写,会出现假满队列的情况

2. 代码如下:

class MyBlockingQueue<E>{//自己实现阻塞队列//有take和putprivate Object object = new Object();private E[] array = (E[]) new Object[50];private int first = 0;//队首private int last = 0;//队尾//先进先出//循环队列//预留一个用来判断是满还是空的内存//first == last是空//(last+1)%array.length == first 是满//出队列public E take(){if(first == last){System.out.println("队列空了");return null;}E value = array[first];first = (first+1)%array.length;//(49+1)%50=0return value;}}//进队列public void put(E value) {if((last+1)%array.length == first){System.out.println("队列满了");return;}array[last] = value;last = (last+1)%array.length;//(49+1)%50 == 0}}}

上述代码实现的是一个普通的简化版循环队列,里面只有put和task方法.我们该怎么优化成带阻塞效果的队列呢?

那就是需要加锁,锁我们应该加在哪里呢?

根据需要我们了解,

• 在take()时如果队列为空的话就进行阻塞,直到有新的元素添加进来,此时解除阻塞效果并将新添加的元素take()出去
• 在put()操作时,如果队列满了的情况下就进行阻塞,直到有元素弹出队列,此时解除阻塞效果并将该元素添加到队尾.

我们通过上述两点需求我们可以这样写:

public E take() throws InterruptedException {//锁对象是this,谁调用这个方法谁就是thissynchronized (this){//如果是空就waitif(first == last){System.out.println("队列空了");this.wait();}E value = array[first];first = (first+1)%array.length;//(49+1)%50=0//唤醒put方法//因为该代码块是加了锁的,所以即使是多线程情况下,当执行完take后,队列也一定不是满的.//此时就可以notify唤醒进行wait()的线程//如果不进行notify就可能会造成put方法一直阻塞下去this.notify();return value;}}//进队列public void put(E value) throws InterruptedException {synchronized (this){if((last+1)%array.length == first){//此时队列满了我们就需要进行阻塞System.out.println("队列满了");this.wait();}array[last] = value;last = (last+1)%array.length;//(49+1)%50 == 0//同上,由于我们的put方法加了锁,所以当进行put之后,该队列一定不是空的//此时便可以唤醒调用take方法的线程this.notify();}}

但是上述代码还有一点点bug,大家看下面的图:

大家都知道,wait()是可以被唤醒的,假如我的代码写的并不严谨,其他的功能就有可能在我wait()的时候提前唤醒我,但是我此时队列还是空的呢,如果此时我take()那么一定会出现异常的.

那这个问题怎么解决?

我们可以改为while()来判断,如果是被其他代码唤醒,那么我还需要再判断队列是否为空,只有满足被唤醒并且队列不为空的情况下才可以继续运行下面的程序…

修改后的代码:

  public E take() throws InterruptedException {synchronized (this){//如果是空就waitwhile(first == last){//用while来判断System.out.println("队列空了");this.wait();}E value = array[first];first = (first+1)%array.length;//(49+1)%50=0//唤醒进队列this.notify();return value;}}//进队列public void put(E value) throws InterruptedException {synchronized (this){while((last+1)%array.length == first){//都用whileSystem.out.println("队列满了");this.wait();}array[last] = value;last = (last+1)%array.length;//(49+1)%50 == 0//释放this.notify();}}}

1.4.2 LinkedBlockingQueue

LinkedBlockingQueue内部使用链表实现的，如果我们不指定它的初始容量，那么它的默认容量就为整形的最大值Integer.MAX_VALUE,由于这个数特别特别的大，所以它也被称为无界队列。

1.4.3 SynchronousQueue

SynchronousQueue最大的不同之处在于，它的容量不同，所以没有地方来暂存元素，导致每次取数据都要先阻塞，直到有数据放入。同理，每次放数据的时候也会阻塞，直到有消费者来取。SynchronousQueue的容量不是1而是0，因为SynchronousQueue不需要去持有元素，它做的就是直接传递。

1.4.4 PriorityBlockingQueue

PriorityBlockingQueue是一个支持优先级的无界阻塞队列，可以通过自定义类实现compareTo()方法来制定元素排序规则，或者初始化时通过构造器参数Comparator来制定排序规则。同时，插入队列的对象必须是可比较大小的，也就是Comparable的，否则就会抛出ClasscastException异常。
它的take方法在队列为空时会阻塞，但是正因为它是无界队列，而且会自动扩容，所以它的队列永远不会满，所以它的put()方法永远不会阻塞，添加操作始终都会成功。

1.5 线程池对于阻塞队列的选择

• FixedThreadPool选取的是LinkedBlcokingQueue（同理SingleThreadExecutor） 首先我们知道LinkedBlockingQueu默认是无限长的，而FixedThreadPool的线程数是固定的，当核心线程数都在被使用时，这个时候如果进来新的任务会被放进阻塞队列中。由于队列是没有容量上限的，队列永远不会被填满，这样就保证了线程池FixedThreadPool和SingleThreadExecutor，不会拒绝新任务的提交，也不会丢失数据。
• CachedThreadPool选取的是SynchronousQueue 首先CachedThreadPool的线程最大数量是无限的，也就意味着它的线程数不会受限制，那么它就不需要额外的空间来存储那些Task，因为每个任务都可以通过新建线程来处理。SynchronousQueue会直接把任务交给线程，不保存它们，效率更好。
• ScheduledThreadPool选取的是延迟队列,对于ScneduledThreadPool而言，它使用的是DelayedWorkQueue，延迟队列的特点是：不是先进先出，而是会按照延迟时间的长短来排序，下一个即将执行的任务会排到队列的最前面。选择使用延迟队列的原因是，ScheduledThreadPool处理的是基于时间而执行的Task，而延迟队列有能力把Task按照执行时间的