龙空技术网

「原创」Java并发编程系列31 | 阻塞队列(上)

Meta多元宇宙 123

前言:

现在你们对“java队列处理高并发”大致比较关切,咱们都想要学习一些“java队列处理高并发”的相关资讯。那么小编同时在网上搜集了一些有关“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:删除数据,当队列为空时,返回;

// 查看元素

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 {

checkNot(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] = ;

if (++takeIndex == items.length)

takeIndex = 0;

count--;

if (itrs != )

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>;

}

put(E e)

获取putLock锁

如果队列满,当前线程阻塞并加入notFull条件等待队列

入队

这个元素入队成功后,队列还没有满,唤醒notFull队列中等待添加元素的线程。(因为添加元素和删除元素不是用的同一个锁导致会有这种情况发生)

释放掉putLock锁

如果添加元素前队列为空,可能会有读线程阻塞,所以在这个元素入队后,就唤醒阻塞的读线程

public void put(E e) throws InterruptedException {

if (e == ) throw new PointerException;

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 = ;

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、数据结构、限流熔断降级等等。

看到这里,证明有所收获

标签: #java队列处理高并发