手记

juc-10-阻塞队列

黑桃SEVEN_PIG
2020-11-28 17:59:33浏览 1092

也许你以前学习数据结构时，接触过Queue(队列)，它具有先进先出的特性。

什么是阻塞队列 BlockingQueue ？
顾名思义它一个队列，具有阻塞功能。

为什么要学习阻塞队列？

  1. 很多朋友可能有些陌生，其实只要使用过线程池，就表示你已经间接地使用过阻塞队列了，阻塞队列是线程池的重要组成部分，后面文章会详细介绍线程池。
  2. 阻塞队列是线程安全的，所以阻塞队列可以用作线程安全的并发容器，比如生产者消费者模式。

1、常见的阻塞队列

阻塞队列关系图

1.1 ArrayBlockingQueue

由数组结构组成的有界阻塞队，需要指定队列容量，可指定公平策略：如果是公平的策略，那么等待了最长时间的线程会优先被处理，不过这会同时带来一定的性能损耗

1.2 LinkedBlockingQueue

由链表结构组成的有界阻塞队列，如果不指定容量，默认为Integer.MAX_VALUE。
内部结构：Node
两把锁Lock：takeLock、putLock
两个条件Condition：notEmpty、notFull

1.3 PriorityBlockingQueue

使用优先级队列实现的无界阻塞队列，支持优先级，自然顺序（而不是先进先出），是 priorityQueue的线程安全版本

1.4 SynchronousQueue

一个不存储元素的阻塞队列，它的容量为 0，每一个put操作都要等待一个take操作。

注意：

  • SynchronousQueue 的容量不是1而是0，因为 SynchronousQueue 不需要去持有元素，它所做的就是直接传递（direct handoff）
  • SynchronousQueue 没有 peek等函数，因为 peek 的含义是取出头节点，但是 SynchronousQueue 的容量为 0，所以连头节点都没有，也就没有 peek 方法。同理，没有 iterate 相关方法。
  • 是一个极好的用来直接传递的并发数据结构
  • SynchronousQueue 是线程池 Executors.newCachedThreadPool() 使用的阻塞队列。

1.5 DelayQueue

使用优先级队列实现的无界阻塞队列，支持延时获取的元素的阻塞队列，元素必须要实现Delayed接口，获取元素剩余的延迟时间
适用场景：订单到期，限时支付等等

2、阻塞队列常用方法

方法 描述
void put(E e) 如果有空间，队尾插入数据；
如果队列已满，则无法插入，阻塞，直到有空间
boolean add(E e) 如果有空间，队尾插入数据，返回true；
如果队列已满，则抛出 IllegalStateException
boolean offer(E e) 如果有空间，队尾插入数据，返回true；
如果队列已满，则返回false
boolean offer
(E e, long timeout,
TimeUnit unit)		 如果有空间，队尾插入数据，返回true；
如果队列已满，等待指定的等待时间，若超时仍未有空间，则返回false
E take() 获取并删除队头节点，
如果队列无数据，则阻塞，直到有数据
boolean remove() 获取并删除队头节点，
如果队列无数据，则抛出 NoSuchElementException
E poll() 获取并删除队头节点，
如果队列无数据，则返回 null
E poll(long timeout,
TimeUnit unit)		 获取并删除队头节点，
如果队列无数据，等待指定的等待时间，若超时仍未有数据，则返回 null
E peek() 获取但是不删除队头节点，如果队列无数据，返回null
E element() 获取但是不删除队头节点，如果队列无数据，抛出 NoSuchElementException

3、ArrayBlockingQueue 的使用和源码

下面以生产者消费者案例为场景，演示 ArrayBlockingQueue 阻塞队列的使用。

3.1 生产者和消费者模式

生产者就是生产数据的线程，消费者就是消费数据的线程。
在多线程开发中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这种生产消费能力不均衡的问题，便有了生产者和消费者模式。
生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

ArrayBlockingQueue实现生产者和消费者模式

public class ArrayBlockingQueueDemo {

    //用阻塞队列作为容器，指定容量为5
    private ArrayBlockingQueue<Goods> blockingQueue = new ArrayBlockingQueue<>(5);

    // 全局商品索引
    static int goodsIndex;

    // 打印日志的lock，是公平锁，等待时间长的线程优先拿到锁
    private Lock fairLock = new ReentrantLock(true);

    public static void main(String[] args) {
        ArrayBlockingQueueDemo demo = new ArrayBlockingQueueDemo();
        for (int i = 0; i < 3; i++) {
            Producer producer = new Producer(demo);
            new Thread(producer, "生产者P" + (i + 1)).start();
        }
        for (int i = 0; i < 2; i++) {
            Consumer consumer = new Consumer(demo);
            new Thread(consumer, "消费者C" + (i + 1)).start();
        }
    }

    /**
     * 生产的商品
     */
    class Goods {
        // 商品下标
        String index;
    }

    /**
     * 生产
     */
    public void produce() throws InterruptedException {
        Goods goods = new Goods();
        // 如果有空间，队尾插入数据；如果队列已满，则无法插入，阻塞，直到有空闲时间
        this.blockingQueue.put(goods);
        this.log(goods, true);//打印生产日志
    }

    /**
     * 消费
     */
    public void consume() throws InterruptedException {
        // 获取并删除队头节点，<br>如果队列无数据，则阻塞，直到有数据
        Goods goods = this.blockingQueue.take();
        this.log(goods, false);//打印消费日志
    }

    /**
     * 打印生产或者消费日志
     *
     * @param goods
     * @param isProduct true表示生产，false表示消费
     */
    private void log(Goods goods, boolean isProduct) {
        this.fairLock.lock();
        try {
            String flag = isProduct ? "生产--" : "消费--";
            if (isProduct) {
                //如果是生产日志，这里设置商品下标
                goods.index = "商品" + goodsIndex++;
            }
            System.out.println(Thread.currentThread().getName() + flag + goods.index);
        } finally {
            this.fairLock.unlock();
        }
    }

}

//生产者
class Producer implements Runnable {
    private ArrayBlockingQueueDemo demo;

    public Producer(ArrayBlockingQueueDemo demo) {
        this.demo = demo;
    }

    @Override
    public void run() {
        while (true) {

            try {
                // 生产者生产
                this.demo.produce();
                TimeUnit.MILLISECONDS.sleep(3000);//模拟生产耗时
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

//消费者
class Consumer implements Runnable {
    private ArrayBlockingQueueDemo demo;

    public Consumer(ArrayBlockingQueueDemo demo) {
        this.demo = demo;
    }

    @Override
    public void run() {
        while (true) {

            try {
                // 消费者消费
                this.demo.consume();
                TimeUnit.MILLISECONDS.sleep(2000);//模拟消费耗时
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

运行结果：

生产者P1生产--商品0
生产者P2生产--商品1
消费者C2消费--商品1
消费者C1消费--商品0
生产者P3生产--商品2
消费者C1消费--商品2
生产者P3生产--商品3
生产者P2生产--商品4
消费者C2消费--商品3
生产者P1生产--商品5
消费者C1消费--商品4
消费者C2消费--商品5
生产者P1生产--商品6
生产者P3生产--商品7
消费者C1消费--商品6
生产者P2生产--商品8
消费者C2消费--商品7
消费者C1消费--商品8
生产者P2生产--商品9
生产者P1生产--商品10
...

3.2 ArrayBlockingQueue 的源码分析

3.2.1 构造器源码

  • ArrayBlockingQueue(int capacity)：指定容量，并使用非公平策略，也就是不指定各个线程对 ArrayBlockingQueue 访问顺序
  • ArrayBlockingQueue(int capacity, boolean fair)：这个是核心构造器，其他构造器都会调用这个构造器；指定容量，指定公平策略，如果是公平的策略，那么等待了最长时间的线程会优先被处理
  • ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c)：指定容量，指定公平策略，把 Collection 中的元素加入 items数组，也就是入队

构造器主要做了下面几个工作：

  • 指定容量
  • 指定公平策略
  • 初始化 Lock 和 Condition

构造器源码：


    //使用数组来存储队列元素
    final Object[] items;
    
    //下一次take, poll, peek or remove操作对应在 items数组下标，也就是队头下标
    int takeIndex;

    //下一次 put, offer, or add 操作对应在 items数组下标，也就是队尾下标
    int putIndex;

    //队列中元素的数量
    int count;

    //用于控制所有访问操作的锁
    final ReentrantLock lock;

    // 队列为空时，控制 take 操作等待的条件
    private final Condition notEmpty;

    // 队列满时，控制 put 操作等待的条件
    private final Condition notFull;

    /**
     * 指定容量，并使用非公平策略，也就是不指定各个线程对 ArrayBlockingQueue 访问顺序
     */
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    /**
     * 指定容量，指定公平策略
     * 如果是公平的策略，那么等待了最长时间的线程会优先被处理
     */
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        // 初始化数组
        this.items = new Object[capacity];
        // 初始化 Lock 和 Condition
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    /**
     * 指定容量，指定公平策略，把 Collection 中的元素加入 items数组，也就是入队
     */
    public ArrayBlockingQueue(int capacity, boolean fair,
                              Collection<? extends E> c) {
        this(capacity, fair);

        final ReentrantLock lock = this.lock;
        lock.lock(); // Lock only for visibility, not mutual exclusion
        try {
            int i = 0;
            try {
                for (E e : c) {
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
                throw new IllegalArgumentException();
            }
            count = i;
            // 如果 Collecion的元素个数超过 items 的容量，则后面加入的元素会覆盖最开始的元素
            putIndex = (i == capacity) ? 0 : i;
        } finally {
            lock.unlock();
        }
    }

3.2.2 put(E e) 源码

void put(E e) ：如果有空间，队尾插入数据；如果队列已满，则无法插入，阻塞，直到有空间。

源码实现：

  • 判断入队元素不为空
  • 获取锁
  • 如果队列已满，notFull.await();释放锁，阻塞当前线程等待
  • 队列不满，执行入队操作，在 putIndex 下标位置，也就是队尾，插入元素，并且执行 notEmpty.signal(); 唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程

put(E e) 源码

    /**
     * 如果有空间，队尾插入数据；如果队列已满，则无法插入，阻塞，直到有空闲时间
     */
    public void put(E e) throws InterruptedException {
        //入队元素不能为空
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //获取锁
        lock.lockInterruptibly();
        try {
            // 如果队列已满，notFull.await();释放锁，阻塞当前线程等待
            while (count == items.length)
                notFull.await();
            // 队列不满，执行入队操作    
            enqueue(e);
        } finally {
            // 释放锁
            lock.unlock();
        }
    }

    //在 putIndex 下标位置，插入元素x，putIndex就是队尾下标
    //并且执行 notEmpty.signal(); 唤醒因为队列为空执行take操作阻塞等待的线程
    //需要获取锁后，才能调用这个方法
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        //在 putIndex 下标位置，插入元素x
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;//当前队列元素+1
        // notEmpty Condition signal，唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程
        notEmpty.signal();
    }

3.2.2 offer(E e)/offer(E e, long timeout, TimeUnit unit) 源码

boolean offer(E e) ：如果有空间，队尾插入数据，返回true；如果队列已满，则返回false
boolean offer(E e, long timeout, TimeUnit unit)：如果有空间，队尾插入数据，返回true；如果队列已满，等待指定的等待时间，若超时仍未有空间，则返回false

源码实现：

  • 判断入队元素不为空
  • 获取锁
  • 如果队列已满， offer(E e) 直接返回false； offer(E e, long timeout, TimeUnit unit)，执行notFull.await(); 释放锁，并且指定当前线程阻塞等待时长，如果等待时间到队列仍然是满的，返回false。
  • 队列不满，执行入队操作，在 putIndex 下标位置，也就是队尾位置，插入元素，并且执行 notEmpty.signal(); 唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程

offer(E e)/offer(E e, long timeout, TimeUnit unit) 源码

    //在 putIndex 下标位置，插入元素x，
    //并且执行 notEmpty.signal(); 唤醒因为队列为空执行take操作阻塞等待的线程
    //需要获取锁后，才能调用这个方法
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        //在 putIndex 下标位置，插入元素x
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;//当前队列元素+1
        // notEmpty Condition signal，唤醒因为 notEmpty.await() 的线程，也就是队列为空执行take操作阻塞等待的线程
        notEmpty.signal();
    }
    
    /**
     * 如果有空间，队尾插入数据，返回true；如果队列已满，则返回false 
     */
    public boolean offer(E e) {
        //入队元素不能为空
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //获取锁
        lock.lock();
        try {
            // 队列已满，返回false
            if (count == items.length)
                return false;
            else {
                // 队列不满，执行入队操作，返回true
                enqueue(e);
                return true;
            }
        } finally {
            //释放锁
            lock.unlock();
        }
    }

    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
        //入队元素不能为空
        checkNotNull(e);
        //等待时长
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        //获取锁
        lock.lockInterruptibly();
        try {
            //队列已满时循环
            while (count == items.length) {
                //等待时间到，返回false
                if (nanos <= 0)
                    return false;
                //notFull.await();释放锁，并且指定当前线程阻塞等待时长 nanos      
                nanos = notFull.awaitNanos(nanos);
            }
            // 队列不满，执行入队操作，返回true
            enqueue(e);
            return true;
        } finally {
            //释放锁
            lock.unlock();
        }
    }

3.2.3 add(E e) 源码

boolean add(E e): 如果有空间，队尾插入数据，返回true；如果队列已满，则抛出 IllegalStateException

源码实现：

  • 调用 offer(e)，如果有空间，队尾插入数据，返回true
  • 如果队列已满 offer(e) 返回false，抛出 IllegalStateException

add(E e) 源码

    //如果有空间，队尾插入数据，返回true；如果队列已满，则抛出 IllegalStateException
    public boolean add(E e) {
        // 调用 offer(e)，如果有空间，队尾插入数据，返回true
        if (offer(e))
            return true;
        else // 如果队列已满 offer(e) 返回false，抛出 IllegalStateException 
            throw new IllegalStateException("Queue full");
    }

3.2.4 take() 源码

E take()： 获取并删除队头节点，如果队列无数据，则阻塞，直到有数据。

源码实现：

  • 如果队列为空，notEmpty.await();释放锁，阻塞当前线程等待
  • 队列不为空，获取并删除 takeIndex 下标位置的元素，也就是队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程

take() 源码

    // 返回并删除 items 数组的takeIndex下标的元素，takeIndex就是队头下标
    // 需要持有锁，才能调用该方法
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        // 获取takeIndex下标的元素
        E x = (E) items[takeIndex];
        // 将该下标的元素设为null，表示删除队头
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        // 队列元素减一    
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // notFull Condition signal，唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程    
        notFull.signal();
        return x;
    }

    // 获取并删除队头节点，如果队列无数据，则阻塞，直到有数据  
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        //获取锁
        try {
            // 队列为空，notEmpty.await();释放锁，阻塞当前线程等待
            while (count == 0)
                notEmpty.await();
            //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程     
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

3.2.5 poll()/poll(long timeout, TimeUnit unit) 源码

E poll()：获取并删除队头节点，如果队列无数据，则返回 null
E poll(long timeout,
TimeUnit unit) ：获取并删除队头节点，如果队列无数据，等待指定的等待时间，若超时仍未有数据，则返回 null

源码实现：

  • 获取锁
  • 如果队列为空， poll() 直接返回null；而 poll(long timeout, TimeUnit unit) 执行notEmpty.await(); 释放锁，并且指定当前线程阻塞等待时长，如果等待时间到队列仍然是空的，返回null。
  • 队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程

poll()/poll(long timeout, TimeUnit unit)源码

    // 返回并删除 items 数组的takeIndex下标的元素，takeIndex就是队头下标
    // 需要持有锁，才能调用该方法
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        // 获取takeIndex下标的元素
        E x = (E) items[takeIndex];
        // 将该下标的元素设为null，表示删除队头
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        // 队列元素减一    
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // notFull Condition signal，唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程    
        notFull.signal();
        return x;
    }

    // 获取并删除队头节点，如果队列无数据，则返回 null
    public E poll() {
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lock();
        try {
            // 如果队列为空，返回null
            //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程 
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }

    // 获取并删除队头节点，如果队列无数据，等待指定的等待时间，若超时仍未有数据，则返回 null 
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        //队列为空时，等待时长
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        // 获取锁
        lock.lockInterruptibly();
        try {
            //队列为空
            while (count == 0) {
                //等待时间到，队列依然为空，返回null
                if (nanos <= 0)
                    return null;
                //notEmpty.await();释放锁，并且指定当前线程阻塞等待时长 nanos      
                nanos = notEmpty.awaitNanos(nanos);
            }
            //队列不为空，获取并删除队头节点，并唤醒因为 notFull.await() 的线程，也就是队列满了，执行put操作阻塞等待的线程 
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

3.2.6 remove() 源码

boolean remove()：获取并删除队头节点，如果队列无数据，则抛出 NoSuchElementException。

源码实现：

  • 调用poll()方法，如果队列不为空，返回队头
  • 如果队列无数据poll()方法返回null，则抛出 NoSuchElementException

remove() 源码

    //获取并删除队头节点，如果队列无数据，则抛出 NoSuchElementException 
    public E remove() {
        // 调用poll()方法：获取并删除队头节点，如果队列无数据，则返回 null
        E x = poll();
        if (x != null) // 队列不为空，返回队头
            return x;
        else // 如果队列无数据，poll()方法返回null，则抛出 NoSuchElementException
            throw new NoSuchElementException();
    }

3.2.7 peek() 源码

E peek()：获取但是不删除队头节点，如果队列无数据，返回null

源码实现：

  • 获取锁
  • 返回items中下标takeIndex的元素，也就是队头节点，如果队列无数据，返回null

peek() 源码

    //获取但是不删除队头节点，如果队列无数据，返回null 
    public E peek() {
        final ReentrantLock lock = this.lock;
        //获取锁
        lock.lock();
        try {
            // 返回items中下标takeIndex的元素，如果队列无数据，返回null 
            return itemAt(takeIndex); 
        } finally {
            lock.unlock();
        }
    }

    /**
     * 返回items中下标i的元素
     */
    @SuppressWarnings("unchecked")
    final E itemAt(int i) {
        return (E) items[i];
    }

3.2.8 element() 源码

E element()：获取但是不删除队头节点，如果队列无数据，抛出 NoSuchElementException

源码实现：

  • 调用 peek() 方法，获取但不删除 items 数组中下标takeIndex的元素，也就是队尾元素，如果队列无数据，peek() 方法返回null
  • 如果 peek() 方法返回null，抛出 NoSuchElementException

element() 源码

    //获取但是不删除队头节点，如果队列无数据，抛出 NoSuchElementException  
    public E element() {
        // 调用 peek() 方法，获取但不删除 items 数组中下标takeIndex的元素，也就是队尾元素，如果队列无数据，返回null 
        E x = peek();
        if (x != null)
            return x;// x!=null,则返回 x
        else // x==null 抛出 NoSuchElementException  
            throw new NoSuchElementException();
    }

小结：
上面简单介绍了常见的阻塞队列，说明了阻塞队列几个常用方法的使用，并以生产者消费者案例为场景，演示 ArrayBlockingQueue 阻塞队列的使用，最后介绍了通过源码分析了 ArrayBlockingQueue 阻塞队列各个核心方法的具体实现步骤。

代码：
github.com/wengxingxia/002juc.git

相关标签
Java
0人推荐
随时随地看视频
慕课网APP

相关阅读

Java队列学习第一篇之列介绍
Java中常用七个阻塞队列的总结
Java 高频面试题：LinkedBlockingQueue 源码解析
突击并发编程JUC系列-阻塞队列 BlockingQueue
java并发编程工具类JUC第一篇:BlockingQueue阻塞队列