码迷,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
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!