码迷，mamicode.com
首页 > 编程语言 > 详细

【原创】Java并发编程系列31 | 阻塞队列（上）

时间：2020-11-24 12:56:45      阅读：11      评论：0      收藏：0      [点我收藏+]

标签：jdb   排序   sig   公众   出队   UNC   tis   ESS   缓存   

【原创】Java并发编程系列31 | 阻塞队列（上）

收录于话题
#并发编程 240 #程序员 2286 #java 976 #进阶架构师 | 并发编程专题 12

★★★建议星标我们★★★
公众号改版后文章乱序推荐，希望你可以点击上方“Java进阶架构师”，点击右上角，将我们设为★“星标”！这样才不会错过每日进阶架构文章呀。

技术图片

2020年Java原创面试题库连载中
【000期】Java最全面试题库思维导图
【001期】JavaSE面试题（一）：面向对象
【002期】JavaSE面试题（二）：基本数据类型与访问修饰符
【003期】JavaSE面试题（三）：JavaSE语法（1）
【004期】JavaSE面试题（四）：JavaSE语法（3）
【005期】JavaSE面试题（五）：String类
【006期】JavaSE面试题（六）：泛型
【007期】JavaSE面试题（七）：异常
【008期】JavaSE面试题（八）：集合之List
【009期】JavaSE面试题（九）：集合之Set
【010期】JavaSE面试题（十）：集合之Map
【011期】JavaSE面试题（十一）：多线程（1）
【012期】JavaSE面试题（十二）：多线程（2）
【013期】JavaSE面试题（十三）：多线程（3）
【014期】JavaSE面试题（十四）：基本IO流
【015期】JavaSE面试题（十五）：网络IO流
【016期】JavaSE面试题（十六）：反射
【017期】JavaSE面试题（十七）：JVM之内存模型
【018期】JavaSE面试题（十八）：JVM之垃圾回收
【020期】JavaSE系列面试题汇总（共18篇）
【019期】JavaWeb面试题（一）：JDBC
【021期】JavaWeb面试题（二）：HTTP协议
【022期】JavaWeb面试题（三）：Cookie和Session
【023期】JavaWeb面试题（四）：JSP
【024期】JavaWeb面试题（五）：Filter和Listener
【025期】Java工具面试题（一）：版本控制工具
【026期】Java工具面试题（二）：项目管理工具
【027期】Java设计模式面试题
【028期】JavaWeb系列面试题汇总（共10篇）
【029期】JavaEE面试题（一）Web应用服务器
【030期】JavaEE面试题（二）SpringMVC
【031期】JavaEE面试题（三）Spring（1）
【032期】JavaEE面试题（四）Spring（2）
【033期】JaveEE面试题（五）MyBatis
【034期】JavaEE面试题（六）Hibernate
【035期】JavaEE面试题（七）SpringBoot（1）
更多内容，点击上面蓝字查看

技术图片
阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。接下来两篇文章就来详细介绍阻塞队列。本文是阻塞队列上篇。
介绍
基本操作
应用
常用阻塞队列及源码 4.1 ArrayBlockingQueue 4.2 LinkedBlockingQueue 4.3 SynchronousQueue 4.4 PriorityBlockingQueue 4.5 DelayQueue

1. 介绍

阻塞队列（BlockingQueue）是一个比普通队列多出两个附加操作的队列。两个操作分别是：
在队列为空时，获取元素的线程会等待队列变为非空。
当队列满时，存储元素的线程会等待队列可用。
阻塞队列（BlockingQueue）被广泛使用在“生产者-消费者”问题中。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。这样可以对各个模块的业务功能进行解耦，生产者将“生产”出来的数据放置在数据容器中，而消费者仅仅只需要在“数据容器”中进行获取数据即可，这样生产者线程和消费者线程就能够进行解耦，只专注于自己的业务功能即可。

2. 基本操作

BlockingQueue基本操作如下：

// 插入元素
add(E e) ：往队列插入数据，当队列满时，插入元素时会抛出IllegalStateException异常；
offer(E e)：当往队列插入数据时，插入成功返回true，否则则返回false。当队列满时不会抛出异常；

// 删除元素
remove(Object o)：从队列中删除数据，成功则返回true，否则为false
poll：删除数据，当队列为空时，返回null；

// 查看元素
element：获取队头元素，如果队列为空时则抛出NoSuchElementException异常；
peek：获取队头元素，如果队列为空则抛出NoSuchElementException异常

// 插入数据：
put()：当阻塞队列容量已经满时，往阻塞队列插入数据的线程会被阻塞，直至阻塞队列已经有空余的容量可供使用；
offer(E e, long timeout, TimeUnit unit)：若阻塞队列已经满时，同样会阻塞插入数据的线程，直至阻塞队列已经有空余的地方，与put方法不同的是，该方法会有一个超时时间，若超过当前给定的超时时间，插入数据的线程会退出；

// 删除数据：
take()：当阻塞队列为空时，获取队头数据的线程会被阻塞；
poll(long timeout, TimeUnit unit)：当阻塞队列为空时，获取数据的线程会被阻塞，另外，如果被阻塞的线程超过了给定的时长，该线程会退出

put(e) 和 take() 是BlockingQueue的核心方法，也是我们比较关注的。

3. 应用

使用普通队列实现生产者-消费者模式，代码如下：

public class BlockingDemo {
    static LinkedList<Integer> queue = new LinkedList<Integer>();
    static int maxSize = 5;

    public static void main(String[] args) throws Exception {
        new Thread("生产者") {
            public void run() {
                while (true) {
                    synchronized (queue) {
                        while (queue.size() >= maxSize) {
                            try {
                                System.out.println("队列满了。。。");
                                queue.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        queue.addLast(1);
                        queue.notify();
                        System.out.println("队列中添加了一个元素, size=" + queue.size());
                    }
                }
            };
        }.start();

        new Thread("消费者") {
            public void run() {
                while (true) {
                    synchronized (queue) {
                        while (queue.size() <= 0) {
                            try {
                                System.out.println("队列空了。。。");
                                queue.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        queue.removeFirst();
                        queue.notify();
                        System.out.println("队列中删除了一个元素, size=" + queue.size());
                    }
                }
            };
        }.start();
    }
}

控制台输出：


队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
队列中删除了一个元素, size=4
队列中删除了一个元素, size=3
队列中删除了一个元素, size=2
队列中删除了一个元素, size=1
队列中删除了一个元素, size=0
队列空了。。。
队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列满了。。。
...部分省略...

使用阻塞队列实现生产者消费者模式：


public class Test {
    static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(5);

    public static void main(String[] args) throws Exception {
        new Thread("生产者") {
            public void run() {
                while (true) {
                    try {
                        queue.put(1);
                        System.out.println("队列中添加了一个元素, size=" + queue.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();

        Thread.sleep(500);

        new Thread("消费者") {
            public void run() {
                while (true) {
                    try {
                        queue.take();
                        System.out.println("队列中删除了一个元素, size=" + queue.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();
    }
}

输出结果如下：


队列中添加了一个元素, size=1
队列中添加了一个元素, size=2
队列中添加了一个元素, size=3
队列中添加了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
队列中添加了一个元素, size=5
队列中删除了一个元素, size=4
...部分省略...

4. 常用阻塞队列

4.1 ArrayBlockingQueue
ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
添加元素时，如果队列满了不能添加元素，就将添加元素的线程阻塞并加入notFull条件队列；当成功删除元素后，队列就可以添加元素了，唤醒notFull条件队列中阻塞的线程，添加元素。
删除元素时，如果队列空了不能删除元素，就将删除元素的线程阻塞并加入notEmpty条件队列；当成功添加元素后，队列就可以删除元素了，唤醒notEmpty条件队列中阻塞的线程，删除元素。
类结构
ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
队列创建时，确定队列大小和是否公平。
源码：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    final Object[] items;// 用于存放元素的数组
    int takeIndex;// 下一次读取操作的位置
    int putIndex;// 下一次写入操作的位置
    int count;// 队列中的元素数量

    // 通过lock及其两个条件notEmpty、notFull控制阻塞
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

    // 创建队列时，确定队列大小和是否公平
    public ArrayBlockingQueue(int capacity, boolean fair) {
 if (capacity <= 0)
  throw new IllegalArgumentException();
 this.items = new Object[capacity];
 lock = new ReentrantLock(fair);
 notEmpty = lock.newCondition();
 notFull =  lock.newCondition();
    }
}

put()
获取锁lock
队列满时，将当前线程加入notFull条件队列阻塞；当有元素出队时，队列就不满了，可以让元素入队了，此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
入队：入队成功之后，唤醒notEmpty条件队列中阻塞的线程，让其元素出队
释放锁lock
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    final Object[] items;// 用于存放元素的数组
    int takeIndex;// 下一次读取操作的位置
    int putIndex;// 下一次写入操作的位置
    int count;// 队列中的元素数量

    // 通过lock及其两个条件notEmpty、notFull控制阻塞
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;
}

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取lock锁
    try {
        /*
         * 队列满时，将当前线程加入notFull条件队列阻塞；
         * 当有元素出队时，队列就不满了，可以让元素入队了，
         * 此时会唤醒notFull条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后将元素入队。
         */
        while (count == items.length)
            notFull.await();
        enqueue(e);// 入队
    } finally {
        lock.unlock();// 解锁
    }
}

private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    // 入队成功之后，唤醒notEmpty条件队列中阻塞的线程，让其元素出队
    notEmpty.signal();
}

take()
获取锁lock
队列空时，将当前线程加入notEmpty条件队列阻塞；当有元素入队时，队列不为空了就可以take出元素，此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
出队：出队成功，唤醒notFull条件队列中阻塞的线程，让其元素入队
释放锁lock

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取lock锁
    try {
        /*
         * 队列空时，将当前线程加入notEmpty条件队列阻塞；
         * 当有元素入队时，队列不为空了就可以take出元素，
         * 此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
         */
        while (count == 0)
            notEmpty.await();
        return dequeue();// 出队
    } finally {
        lock.unlock();// 解锁
    }
}

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 出队成功，唤醒notFull条件队列中阻塞的线程，让其元素入队
    notFull.signal();
    return x;
}

4.2 LinkedBlockingQueue
LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）

  • 锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
  • 锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
    ArrayBlockingQueue的读写使用同一个锁来保证数据安全。LinkedBlockingQueue的读写分别用不同的锁来保证数据安全，采用不同的锁可以使读线程和写线程并发执行，提高了吞吐量，但也增加了编程的复杂度。
    类结构
    LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）
    锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
    锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
// 节点类 单向链表
static class Node<E> {
 E item;
 Node<E> next;
 Node(E x) { item = x; }
}

private final int capacity;// 队列容量 不设置默认为Integer.MAX_VALUE
private final AtomicInteger count = new AtomicInteger(0);// 队列中的元素数量
private transient Node<E> head;// 队头
private transient Node<E> last;// 队尾

// take, poll, peek 等读操作的方法需要获取到这个锁
private final ReentrantLock takeLock = new ReentrantLock();
// 如果读操作的时候队列是空的，加入notEmpty等待队列
private final Condition notEmpty = takeLock.newCondition();

// put, offer 等写操作的方法需要获取到这个锁
private final ReentrantLock putLock = new ReentrantLock();
// 如果写操作的时候队列是满的，加入notFull等待队列
private final Condition notFull = putLock.newCondition();

// 有界队列， 不设置容量，默认为Integer.MAX_VALUE
public LinkedBlockingQueue() {
 this(Integer.MAX_VALUE);
}

public LinkedBlockingQueue(int capacity) {
 if (capacity <= 0) throw new IllegalArgumentException();
 this.capacity = capacity;
 last = head = new Node<E>(null);
}

put(E e)
获取putLock锁
如果队列满，当前线程阻塞并加入notFull条件等待队列
入队
这个元素入队成功后，队列还没有满，唤醒notFull队列中等待添加元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
释放掉putLock锁
如果添加元素前队列为空，可能会有读线程阻塞，所以在这个元素入队后，就唤醒阻塞的读线程

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();// 获取putLock锁
    try {
        // 如果队列满，当前线程阻塞并加入notFull条件等待队列
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);// 入队
        c = count.getAndIncrement();// count 原子加 1，注意：这里返回的c是原来的值，并不是加1后的值。
        /*
         * 这个元素入队成功后，队列还没有满，notFull.signal() 唤醒notFull队列中等待添加元素的线程。
         * 为什么队列还没有满，但是添加元素线程却在阻塞状态呢？
         * 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
         * 当添加元素时发现队列满了，线程阻塞。此时另一个线程执行删除操作，队列又不满了。
         * 于是出现了这个情况：队列还没有满，但是添加元素线程却在阻塞状态。
         */
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();// 入队后，释放掉 putLock
    }
    /*
     * c == 0表示队列在这个元素入队前是空的，队列为空时可能会有读线程阻塞
     * 所以在这个元素入队后，就唤醒阻塞的读线程
     */
    if (c == 0)
        signalNotEmpty();
}

/**
 * 队列尾部插入元素
 */
private void enqueue(Node<E> node) {
    last = last.next = node;
}

/**
 * 唤醒读线程
 */
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

take()
获取takeLock锁
如果队列空，当前线程阻塞并加入notEmpty条件等待队列
出队
这个元素出队成功后，队列还有元素，唤醒notEmpty队列中等待删除元素的线程。（因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生）
释放takeLock锁
如果添加元素前队列是满的，可能有写线程阻塞等待，所以唤醒写线程

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();// 获取锁takeLock
    try {
        // 如果队列为空，当前线程阻塞并加入notEmpty条件等待队列
        while (count.get() == 0) {
            notEmpty.await();
        }
        x = dequeue();// 出队
        c = count.getAndDecrement();// count 进行原子减 1,注意：这里返回的c是原来的值，并不是减1后的值。
        /*
         * 这个元素出队成功后，队列还没有满，notEmpty.signal() 唤醒notEmpty队列中等待删除元素的线程。
         * 当队列中还有元素时，为什么会有读线程在阻塞呢？
         * 因为添加元素和删除元素不是用的同一个锁，所以添加元素和删除元素是可以同时进行的。
         * 当删除元素时发现队列空了，线程阻塞。此时另一个线程执行添加操作，队列又不空了。
         * 于是出现了这个情况：当队列中还有元素时，会有读线程在阻塞状态。
         */
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();// 释放锁takeLock
    }
    // c == capacity表示删除元素之前队列是满的，队列满时可能有写线程阻塞等待，所以唤醒写线程
    if (c == capacity)
        signalNotFull();
    return x;
}
// 出队
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

// 唤醒写线程来写
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

并发系列文章汇总

【原创】01|开篇获奖感言
【原创】02|并发编程三大核心问题
【原创】03|重排序-可见性和有序性问题根源
【原创】04|Java 内存模型详解
【原创】05|深入理解 volatile
【原创】06|你不知道的 final
【原创】07|synchronized 原理
【原创】08|synchronized 锁优化
【原创】09|基础干货
【原创】10|线程状态
【原创】11|线程调度
【原创】12|揭秘 CAS
【原创】13|LockSupport
【原创】14|AQS 源码分析
【原创】15|重入锁 ReentrantLock
【原创】16|公平锁与非公平锁
【原创】17|读写锁八讲（上）
【原创】18|读写锁八讲（下）
【原创】19|JDK8新增锁StampedLock
【原创】20|StampedLock源码解析
【原创】21|Condition-Lock的等待通知
【原创】22|倒计时器CountDownLatch
【原创】22|倒计时器CountDownLatch
【原创】23|循环屏障CyclicBarrier
【原创】24|信号量Semaphore
【原创】25|交换器Exchangere
【原创】26|ConcurrentHashMap（上）
【原创】27|ConcurrentHashMap（下）
【原创】28|Copy-On-Write容器
【原创】29|ConcurrentLinkedQueue

之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。
技术图片
获取方式：点“在看”，V信关注上述Java最全面试题库号并回复 【面试】即可领取，更多精彩陆续奉上。

看到这里，证明有所收获
必须点个在看支持呀，喵

【原创】Java并发编程系列31 | 阻塞队列（上）

标签：jdb   排序   sig   公众   出队   UNC   tis   ESS   缓存   

原文地址：https://blog.51cto.com/15009303/2552577
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！