前言:
现在你们对“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队列处理高并发