阻塞队列知道吗

Posted 2021-03-26 半糖君

 

一，阻塞队列？

     当阻塞队列为空时，获取（take）操作是阻塞的；当阻塞队列为满时，添加（put）操作是阻塞的。

　　　　　　

 

 

 

  二，为什么用，有什么好处？

         阻塞队列不用手动控制什么时候该被阻塞，什么时候该被唤醒，简化了操作。

• 在多线程领域：所谓阻塞，在某些情况下会挂起线程(即阻塞)，一旦条件满足，被挂起的线程又会自动被唤醒
• 好处是我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切BlockingQueue都一手包办了；在concurrent 包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，会给我们程序带来不小的复杂度

三，架构梳理与种类分析

 

 

 （1）种类分析

•  ArrayBlockingQueue：由数组结构组成的有界阻塞队列。
• LinkedBlockingQueue：由链表结构组成的有界(大小默认为Integer.MAX_VALUE)阻塞队列。
• PriorityBlockingQueue：支持优先级排序的无界阻塞队列。
• DelayBlockingQueue：使用优先级队列实现的延迟无界阻塞队列。
• SynchronousQueue：不存储元素的阻塞队列，也即单个元素的队列。
• LinkedTransferQueue：由链表结构组成的无界阻塞队列。
• LinkedBlockingDeque：由链表结构组成的双向阻塞队列。

    注意：

1.  粗体标记的三个用得比较多，许多消息中间件底层就是用它们实现的。
2. 需要注意的是LinkedBlockingQueue虽然是有界的，但有个巨坑，其默认大小是Integer.MAX_VALUE，高达21亿，一般情况下内存早爆了（在线程池的ThreadPoolExecutor有体现）。
3. API：抛出异常是指当队列满时，再次插入会抛出异常；返回布尔是指当队列满时，再次插入会返回false；阻塞是指当队列满时，再次插入会被阻塞，直到队列取出一个元素，才能插入。超时是指当一个时限过后，才会插入或者取出。API使用见BlockingQueueDemo。
4. SynchronousQueue 没有容量。与其他BlockingQueue不同，SynchronousQueue 是一个不存储元素的 BlockingQueue。每一个put操作必须等待一个take操作，否则不能继续添加元素，反之亦然。

 

（2）BlockingQueue的核心方法

       

•  抛出异常

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

•  特殊值

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

•  一直阻塞

- 当阻塞队列满时，生产者线程继续往队列里put元素，队列会一直阻塞生产线程知道put数据或者响应中断退出
- 当阻塞队列空时，消费者线程试图从队列里take元素，队列会一直阻塞消费者线程知道队列可用

•  超时退出

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

由于Java中的阻塞队列接口BlockingQueue继承自Queue接口，因此先来看看阻塞队列接口为我们提供的主要方法

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

    //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量）
    //在成功时返回 true，如果此队列已满，则抛IllegalStateException。 
    boolean add(E e); 
    
    //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量） 
    // 将指定的元素插入此队列的尾部，如果该队列已满， 
    //则在到达指定的等待时间之前等待可用的空间,该方法可中断 
    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; 
    
    //将指定的元素插入此队列的尾部，如果该队列已满，则一直等到（阻塞）。 
    void put(E e) throws InterruptedException; 
    
    //获取并移除此队列的头部，如果没有元素则等待（阻塞）， 
    //直到有元素将唤醒等待线程执行该操作 
    E take() throws InterruptedException; 
    
    //获取并移除此队列的头部，在指定的等待时间前一直等到获取元素， //超过时间方法将结束
    E poll(long timeout, TimeUnit unit) throws InterruptedException; 
    
    //从此队列中移除指定元素的单个实例（如果存在）。 
    boolean remove(Object o); 
}

    //除了上述方法还有继承自Queue接口的方法 
    //获取但不移除此队列的头元素,没有则跑异常NoSuchElementException 
    E element(); 
    
    //获取但不移除此队列的头；如果此队列为空，则返回 null。 
    E peek(); 
    
    //获取并移除此队列的头，如果此队列为空，则返回 null。 
    E poll();

这里我们把上述操作进行分类

插入方法：

add(E e) : 添加成功返回true，失败抛IllegalStateException异常
offer(E e) : 成功返回 true，如果此队列已满，则返回 false。
put(E e) :将元素插入此队列的尾部，如果该队列已满，则一直阻塞
删除方法:

remove(Object o) :移除指定元素,成功返回true，失败返回false
poll() : 获取并移除此队列的头元素，若队列为空，则返回 null
take()：获取并移除此队列头元素，若没有元素则一直阻塞。
检查方法

element() ：获取但不移除此队列的头元素，没有元素则抛异常
peek() :获取但不移除此队列的头；若队列为空，则返回 null。

 

四，应用场景：

• 线程通信之生产消费者（传统版生产消费，阻塞队列版）
• 线程池
• 消息中间件

过度期间：

    

题目：synchronized 和 Lock 有什么区别？用新的Lock有什么好处？

1. 原始构成

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

　　2，使用方法

synchronized 不需要用户去手动释放锁，当 synchronized 代码执行完成后系统会自动让线程释放对锁的占用
ReentrantLock 则需要用户去手动释放锁，若没有主动释放锁，就有可能导致出现死锁现象。需要 lock()和unLock()方法配置tru/finally语句块来完成
　　3，等待是否可断

synchronized 不可中断，除非抛出异常或者正常运行完成
ReentrantLock 可中断，a.设置超时方法 tryLock(long timeout,TimeUnit unit)
b.lockInterruptibly() 放代码块，调用 interrupt() 方法可中断
　　4，加锁是否公平

synchronized 非公平锁
ReentrantLock 两者都可以，默认非公平锁，构造方法可以传入boolean值，true为公平锁，false为非公平锁
　　5，锁绑定多个条件Condition

synchronized 没有
ReentrantLock 用来实现分组唤醒需要唤醒的线程们，可以精确唤醒，而不是像synchronized要么随机唤醒一个线程要么唤醒全部线程。

 

 

     举例：锁绑定多个条件Condition

public class SyncAndReetrantLockDemo {

    public static void main(String[] args) {

        /*
            多线程之间按顺序调用，实现A->B->C三个线程启动，要求如下：
            AA打印5次，BB打印10次，CC打印15次...共10轮
         */
        ShareResource shareResource = new ShareResource();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                shareResource.print(5,shareResource.getC1(),shareResource.getC2(),2);
            }
        }, "A").start();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                shareResource.print(10,shareResource.getC2(),shareResource.getC3(),3);
            }
        }, "B").start();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                shareResource.print(15,shareResource.getC3(),shareResource.getC1(),1);
            }
        }, "C").start();

    }
}

class ShareResource{

    private int number = 1;//A:1 B:2 C:3
    private Lock lock = new ReentrantLock();
    private Condition c1 = lock.newCondition();
    private Condition c2 = lock.newCondition();
    private Condition c3 = lock.newCondition();

    public Condition getC1() {
        return c1;
    }

    public Condition getC2() {
        return c2;
    }

    public Condition getC3() {
        return c3;
    }

    public void print(int times, Condition condition1, Condition condition2, int num){
        lock.lock();
        try {
            int temp = num == 1 ? 3 : (num-1);
            while (number != temp){
                condition1.await();
            }
            for (int i = 0; i < times; i++) {
                System.out.println(Thread.currentThread().getName() + " " + i);
            }
            //通知标志
            this.number = num;
            condition2.signal();
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

• 生产者消费者队列版

public class ProdConsumer_BlockQueueDemo {

    public static void main(String[] args) throws Exception {
        MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + " 生产线程启动");
            try {
                myResource.myProd();
                System.out.println();
                System.out.println();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "Prod").start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + " 消费线程启动");
            try {
                myResource.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "Consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("5秒钟时间到，main线程叫停，活动结束");
        myResource.stop();
    }
}

class MyResource{

    private volatile boolean flag = true;//默认开启，进行生产消费
    private AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<String> blockingQueue = null;

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void myProd() throws Exception {
        String data = null;
        boolean retValue;

        while (flag) {
            data = atomicInteger.incrementAndGet() + "";
            retValue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
            if (retValue){
                System.out.println(Thread.currentThread().getName() + " 插入队列" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + " 插入队列" + data + "失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName() + " 被叫停了！表示flag=false，生产动作结束");
    }

    public void myConsumer() throws Exception {
        String result;
        while (flag) {
            result = blockingQueue.poll(2L, TimeUnit.SECONDS);
            if (null == result || result.equalsIgnoreCase("")){
                flag = false;
                System.out.println(Thread.currentThread().getName() + " 超过2秒钟没有取到，消费队列退出");
                return;
            }
            System.out.println(Thread.currentThread().getName() + " 消费队列" + result + "成功");
        }
    }

    public void stop() throws Exception {
        this.flag = false;
    }

}

 参考博主：https://blog.csdn.net/javazejian/article/details/77410889