【面试专栏】阻塞队列

文章同步更新在个人公众号“梓莘”，欢迎大家关注，相互交流。

阻塞队列

ArrayBlockingQueue 是一个基于数组的有界阻塞队列，此队列基按FIFO原则对元素进行排序

LinkedBlockQueue:一个基于链表结构的阻塞队列，次队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue

SynchromousQueue:一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockQueue

阻塞队列：首先它是一个队列，而一个阻塞队列在数据结构中所其的作用大致如图

在这里插入图片描述

当阻塞队列是空时，从队列中获取元素的操作将会被阻塞

当阻塞队列是满时，往队列中添加元素的操作将会被阻塞

试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程王空的队列中插入新的元素。

试图往满的阻塞队列中添加新元素的线程通用也会被阻塞，直到其他的线程从队列中移除一个或者多个元素或者完全清空队列后使队列重新变得空闲起来并后续新增。

为什么用 好处是什么？

在多线程领域中：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒。

为什么需要BlockingQueue?

好处是我们不需要关系什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切BlockingQueue都实现了。

在concurrent包发布之前，在多线程环境下，我们每个线程都必须自己去控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不少的复杂度。

BlockingQueue种类

1. ArrayBlockingQueue:由数组结果组成的有界阻塞队列
2. LinkedBlockingQueue:由链表结构组成的有界阻塞队列，默认大小为Integer.MAX_VALUE
3. PriorityBlockingQueue :支持优先级排序的无界阻塞队列
4. DelayQueue: 使用优先级队列实现的延迟无界阻塞队列
5. SynchronousQueue:不存储元素的阻塞队列，也就是单个元素的队列
6. LinkedTransferQueue:由链表结构组成的无界阻塞队列
7. LinkedBlockingQueue:由链表结构组成的双向阻塞队列

阻塞队列的核心方法

在这里插入图片描述

• 抛出异常：当阻塞队列满时，再往队列里add插入元素会抛出java.lang.IllegalStateException: Queue full
  当阻塞队列空时，再从队列里remove移除元素会抛NoSuchElementException

public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
        System.out.println(blockingQueue.add("a"));
        System.out.println(blockingQueue.add("b"));
        System.out.println(blockingQueue.add("c"));
        System.out.println(blockingQueue.element());
        //java.lang.IllegalStateException: Queue full
        //System.out.println(blockingQueue.add("d"));
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        //java.util.NoSuchElementException
        //System.out.println(blockingQueue.remove());
    }

• 特殊值：插入方法，成功true失败false
  移除方法，成功返回出队列的元素，队列里面没有就返回null

  public static void main(String[] args) {
       BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
       System.out.println(blockingQueue.offer("a"));
       System.out.println(blockingQueue.offer("b"));
       System.out.println(blockingQueue.offer("c"));
       //false
       System.out.println(blockingQueue.offer("d"));
       System.out.println(blockingQueue.peek());
       System.out.println(blockingQueue.poll());
       System.out.println(blockingQueue.poll());
       System.out.println(blockingQueue.poll());
       //null
       System.out.println(blockingQueue.poll());
   }

• 一直阻塞：当阻塞队列满时，生产者线程继续往队列中put元素，队列会一直阻塞生产线程直到put数据或者响应中断请求
  当队列空时，消费者线程视图从队列里take元素，队列会一直阻塞消费者线程直到队列可用

public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        //blockingQueue.put("d");
        blockingQueue.take();
        blockingQueue.take();
        blockingQueue.take();
        //blockingQueue.take();
    }

• 超时退出：当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程会退出

 public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
        blockingQueue.offer("a",2L, TimeUnit.SECONDS);
        blockingQueue.offer("b",2L, TimeUnit.SECONDS);
        blockingQueue.offer("c",2L, TimeUnit.SECONDS);
        blockingQueue.offer("d",2L, TimeUnit.SECONDS);
    }

SynchronousQueue

SynchronousQueue与其他BlockingQueue不同，SynchronousQueue是一个不存储元素的BlockingQueue,每一个put操作必须要等待一个take操作，否则不能继续添加元素，反之亦然。

package com.zixin;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

/**
 * @ClassName SynchronousQueueDemo
 * @Description 
 * @Author zishen
 * @Date 2019/12/31 9:10
 * @Version 1.0
 * AA put 1
 * BB take 1
 * AA put 2
 * BB take 2
 * AA put 3
 * BB take 3
 **/
public class SynchronousQueueDemo {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" put 1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+" put 2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName()+" put 3");
                blockingQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"AA").start();

        new Thread(()->{
            try {
                Thread.sleep(5);
                System.out.println(Thread.currentThread().getName()+" take 1");
                blockingQueue.take();
                Thread.sleep(5);
                System.out.println(Thread.currentThread().getName()+" take 2");
                blockingQueue.take();
                Thread.sleep(5);
                System.out.println(Thread.currentThread().getName()+" take 3");
                blockingQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"BB").start();
    }
}

线程消费之生产者消费者

package com.zixin;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 资源类
 */
class ShareData{
private int number = 0;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
public  void increment()throws Exception{
    lock.lock();
    try{
    //1、判断
    while(number !=0){
        //等待 不能生产
        condition.await();
    }
    //2、干活
    number++;
    System.out.println(Thread.currentThread().getName()+"  "+number);
    condition.signalAll();
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        lock.unlock();
    }
}

    public  void decrement()throws Exception{
        lock.lock();
        try{
            //1、判断  要使用while 如果用if可能会产生虚假唤起
            while(number ==0){
                //等待 不能生产
                condition.await();
            }
            //2、干活
            number--;
            System.out.println(Thread.currentThread().getName()+"  "+number);
            condition.signalAll();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}
/**
 * @ClassName ProduceConsumerTraditionDemo
 * @Description 初始值为0的变量。两个线程对其交替操作，一个加1一个减1 来5轮
 *     1、线程  操作   资源类
 *     2、判断  干活   通知
 *     3、防止虚假唤醒操作
 * @Author zixin
 * @Date 2019/12/31 10:35
 * @Version 1.0
 **/
public class ProduceConsumerTraditionDemo {

    public static void main(String[] args) {

        ShareData shareData = new ShareData();
        new Thread(()->{
            for (int i=0;i<5;i++){
                try {
                    shareData.increment();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },"AA").start();
        new Thread(()->{
            for (int i=0;i<5;i++){
                try {
                    shareData.decrement();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },"BB").start();
    }
}

synchronized和lock的区别

• 原始构成：
  synchronized是关键字属于JVM层面
  monitorenter(底层是通过monitor对象来完成，其实wait/notify等方法也依赖于monitor，对象只有在同步块或方法中才能调wait/notify方法)
  monitorexit
  lock是具体类（java.util.concurrent.locks.Lock）是api层面的锁

• 使用方法：
  snchronized不需要用户去手动释放锁，当synchronized代码执行完成后系统会自动让线程释放对锁的占用
  ReentrantLock则需要用户去手动释放锁若么有主动释放锁，就有可能导致出现死锁现象
  需要lock()和unlock()方法配合try/finally语句块来完成

• 等待是否可中断：
  synchronized不可中断，除非抛出异常或正常运行完成。
  ReentrantLock可中断：
  1、设置超时方法tryLock(long timeout,TimeUnit unit)
  2、LockInterruptibly()放代码块中，调用interrupt()方法可中断

• 加锁是否公平：
  synchronized非公平锁
  ReentrantLock两者都可以，默认公平锁，构造方法可以传入boolean值true为公平锁，false为非公平锁

• 锁绑定多个条件Condition
  synchronized没有
  ReentrantLock用来实现介绍唤醒需要唤醒的线程组，可以精确唤醒，而不是像synchronized要么唤醒一个线程要么唤醒全部线程