BlockingQueue队列学习

Posted 2020-07-03 逐客教我

           今天看了下BlockingQueue的几种实现，记录下以便以后复习。

           首先来看一下BlockingQueue的家族成员：

          BlockingQueue除了先进先出外，还有两个操作：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。           

阻塞队列提供了四种处理方法:

 

方法\\处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

• 抛出异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出IllegalStateException(“Queue full”)异常。当队列为空时，从队列里获取元素时会抛出NoSuchElementException异常 。
• 返回特殊值：插入方法会返回是否成功，成功则返回true。移除方法，则是从队列里拿出一个元素，如果没有则返回null
• 一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里take元素，队列也会阻塞消费者线程，直到队列可用。
• 超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。

 

 

         ArrayBlockingQueue一个由数组支持的有界的阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。实现原理使用ReentrantLock和Condition。下边写个例子：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/*
* 现有的程序代码模拟产生了16个日志对象，并且需要运行16秒才能打印完这些日志，
* 请在程序中增加4个线程去调用parseLog()方法来分头打印这16个日志对象，
* 程序只需要运行4秒即可打印完这些日志对象。
*/
public class PrintLogTest {
    
           public static void main(String[] args) throws Exception {
                 // 新建一个等待队列
                 final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(16);
                 // 四个线程
                 for (int i = 0; i < 4; i++) {
                       new Thread(new Runnable() {
                             @Override
                             public void run() {
                                   while (true ) {
                                         try {
                                              String log = (String) bq.take();
                                               parseLog(log);
                                        } catch (Exception e) {
                                        }
                                  }
                            }
                      }).start();
                }
                 for (int i = 0; i < 16; i++) {
                      String log = (i + 1) +  "–> ";
                      bq.put(log); // 将数据存到队列里！
                }
          }
           // parseLog方法内部的代码不能改动
           public static void parseLog(String log) {
                System. out.println(log + System.currentTimeMillis());
                 try {
                      Thread. sleep(1000);
                } catch (InterruptedException e) {
                      e.printStackTrace();
                }
          }
}

            LinkedBlockingQueue ： 基于链表的阻塞队列，同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列 中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时 （LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反 之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别 采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

    /** Lock held by take, poll, etc */

    private final ReentrantLock takeLock = new ReentrantLock();

    /** Lock held by put, offer, etc */

    private final ReentrantLock putLock = new ReentrantLock();

            

 

              PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定），但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。因此使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁。

           

              DelayQueue：是一个支持延时获取元素的使用优先级队列的实现的无界阻塞队列。队列中的元素必须实现Delayed接口和Comparable接口，也就是说DelayQueue里面的元素必须有public int compareTo( T o)和long getDelay(TimeUnit unit)方法存在，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。我们可以将DelayQueue运用在以下应用场景：

• 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。
• 定时任务调度。使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，从比如TimerQueue就是使用DelayQueue实现的。

       

(1)实现一个Student对象作为DelayQueue的元素必须实现Delayed 接口的两个方法；

import java.util.concurrent.Delayed;

import java.util.concurrent.TimeUnit;

public class Student implements Delayed { //必须实现Delayed接口

       private String name ;

       private long submitTime ;// 交卷时间

       private long workTime ;// 考试时间

       public String getName() {

             return this .name + ” 交卷,用时” + workTime;

      }

       public Student(String name, long submitTime) {

             this.name = name;

             this.workTime = submitTime;

             this.submitTime = TimeUnit.NANOSECONDS.convert(submitTime, TimeUnit.MILLISECONDS ) + System.nanoTime ();

            System. out.println(this.name + ” 交卷,用时” + workTime);

      }

       //必须实现getDelay方法

       public long getDelay(TimeUnit unit) {

//          返回一个延迟时间

             return unit.convert(submitTime – System.nanoTime (), unit.NANOSECONDS );

      }

       //必须实现compareTo方法

       public int compareTo(Delayed o) {

//          比较的方法

            Student that = (Student) o;

             return submitTime > that.submitTime ? 1 : ( submitTime < that.submitTime ? -1 : 0);

      }

}

(2)执行运行类如下：

package demo.thread;

import java.util.concurrent.DelayQueue;

public class DelayQueueTest {

       public static void main(String[] args) throws Exception {

             // 新建一个等待队列

             final DelayQueue<Student> bq = new DelayQueue<Student>();

             for (int i = 0; i < 5; i++) {

                  Student student = new Student(“学生” +i,Math.round((Math. random()*10+i)));

                  bq.put(student); // 将数据存到队列里！

            }

             //获取但不移除此队列的头部；如果此队列为空，则返回 null。

            System. out.println(“bq.peek()”+bq.peek().getName());

             //获取并移除此队列的头部，在可从此队列获得到期延迟的元素，或者到达指定的等待时间之前一直等待（如有必要）。

             //poll(long timeout, TimeUnit unit) 大家可以试一试这个方法

      }

}

运行结果如下：每次运行结果都不一样，一问，我们获得永远是队列里面的第一个元素；

学生0 交卷,用时8

学生1 交卷,用时6

学生2 交卷,用时10

学生3 交卷,用时10

学生4 交卷,用时9

bq.peek()学生1 交卷,用时6

可以慢慢的在以后的工作当中体会DelayQueue的用法。