java之阻塞队列BlockingQueue解析
更新时间：2023年10月24日 10:18:11 作者：强强爱java
这篇文章主要介绍了java之阻塞队列BlockingQueue解析,阻塞队列是一种线程安全的数据结构,用于在多线程环境下进行数据交换,它提供了一种阻塞的机制,当队列为空时,消费者线程将被阻塞,直到队列中有数据可供消费,需要的朋友可以参考下
一、阻塞队列概念
阻塞队列，顾名思义，首先它是一个队列（先进先出），而一个阻塞队列在数据结构所起到的作用大致如下图：
- 线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素
- 当阻塞队列是空是，从队列中获取元素的操作会被阻塞
- 当阻塞队列是满时，从队列中添加元素的操作会被阻塞
- 试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。
- 试图网已满的阻塞队列中添加新元素的线程同样会被阻塞，直到其他的线程从列中移除一个或者多个元素或者完全清空队列后使队列重新变得空闲起来并后续新增
二、阻塞队列的好处
1.在多线程领域：所谓阻塞，在某些情况下会挂起线程，一旦满足条件，被挂起的线程又会自动被唤醒。
2.我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切BlockingQueue（阻塞队列）都给你一手包办了
PS：在concurrent包发布以前，在多线程环境下，我们每个程序员都必须自己控制这些细节，尤其还要兼顾效率和线程安全，而这回给我们程序带来不小的复杂度
三、阻塞队列种类
- ArrayBlockingQueue：由数据结构组成的有界阻塞队列
- LinkedBlockingQueue：由链表结构组成的有界（但大小默认值为 Integer.MAX_VALUE )阻塞队列
- PriorityBlockingQueue：支持优先级排序的无界阻塞队列
- DelayQueue：使用优先级队列实现的延迟无界阻塞队列
- SynchronousQueue：不存储元素的阻塞队列，也即单个元素的队列
- LinkedTransferQueue：由链表结构组成的无界阻塞队列
- LinkedBlockingDeque：由历览表结构组成的双向阻塞队列
PS：重点掌握ArrayBlockingQueue、LinkedBlockingQueue、SychronousQueue三种
四、BlockingQueue的核心方法
| 方法类型 | 抛出异常 | 特殊值 | 一直阻塞 | 超时退出 |
| 插入 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除 | remove() | poll() | take | poll(time,unit) |
| 检查 | element() | peek() | 不可用 | 不可用 |
说明：
| 方法类型 | 情况 |
| 抛出异常 | 当阻塞队列满时，再往队列中add会抛 IllegalStateException: Queue full； 当阻塞队列空时，再从队列里remove会抛 NoSuchElementException |
| 特殊值 | offer(e)插入方法，成功true失败false poll() 移除方法，成功返回出队列的元素，队列里没有就返回null |
| 一直阻塞 | 当阻塞队列满时，生产者线程继续往队列里put元素，队列会一直阻塞线程直到take数据或响应中断退出 当阻塞队列空时，消费者线程试图从队列take元素，队列会一直阻塞消费者线程直到队列可用 |
| 超时退出 | 当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程会退出 |
五、示例代码
1.使用SychronousQueue队列
package com.jian8.juc.queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
/*
* 阻塞队列SynchronousQueue演示
* */
public class SynchronousQueueDemo {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + "\t put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName() + "\t put 2");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName() + "\t put 3");
blockingQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "AAA").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "BBB").start();
}
}
运行结果：
2.传统版生产者消费者模式
package com.jian8.juc.queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 一个初始值为零的变量，两个线程对其交替操作，一个加1一个减1，来5轮
* 1. 线程 操作 资源类
* 2. 判断 干活 通知
* 3. 防止虚假唤起机制
*/
public class ProdConsumer_TraditionDemo {
public static void main(String[] args) {
ShareData shareData = new ShareData();
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}, "ProductorA " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}, "ConsumerA " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}, "ProductorB " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}, "ConsumerB " + i).start();
}
}
}
//资源类
class ShareData {
private int number = 0;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
//生产
public void increment() throws Exception {
lock.lock();
try {
//1.判断
while (number != 0) {
//等待中，不能生产
condition.await();
}
//2.干活
number++;
System.out.println(Thread.currentThread().getName() + "\t" + number);
//3.通知
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
//消费
public void decrement() throws Exception {
lock.lock();
try {
//1.判断
while (number == 0) {
//等待中，不能消费
condition.await();
}
//2.消费
number--;
System.out.println(Thread.currentThread().getName() + "\t" + number);
//3.通知
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
运行结果（部分）：
3.阻塞队列版（以ArrayBlockingQueue为例）生产者消费者模式
package com.jian8.juc.queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class ProdConsumer_BlockQueueDemo {
public static void main(String[] args) {
MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "\t生产线程启动");
try {
myResource.myProd();
} catch (Exception e) {
e.printStackTrace();
}
}, "Prod").start();
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "\t消费线程启动");
try {
myResource.myConsumer();
} catch (Exception e) {
e.printStackTrace();
}
}, "Consumer").start();
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("5s后main叫停，线程结束");
try {
myResource.stop();
} catch (Exception e) {
e.printStackTrace();
}
}
}
//资源类
class MyResource {
private volatile boolean flag = true;//默认开启，进行“生产+消费”活动
private AtomicInteger atomicInteger = new AtomicInteger();
BlockingQueue<String> blockingQueue = null;//消息队列
public MyResource(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
System.out.println(blockingQueue.getClass().getName());
}
//生产
public void myProd() throws Exception {
String data = null;
boolean retValue;
while (flag) {
data = atomicInteger.incrementAndGet() + "";
//超过2s没生产成功，退出生产
retValue = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
if (retValue) {
System.out.println(Thread.currentThread().getName() + "\t插入队列" + data + "成功");
} else {
System.out.println(Thread.currentThread().getName() + "\t插入队列" + data + "失败");
}
TimeUnit.SECONDS.sleep(1);
}
System.out.println(Thread.currentThread().getName() + "\t大老板叫停了，flag=false，生产结束");
}
//消费
public void myConsumer() throws Exception {
String result = null;
while (flag) {
//超过2s没从队列获取数据，消费退出
result = blockingQueue.poll(2, TimeUnit.SECONDS);
if (null == result || result.equalsIgnoreCase("")) {
flag = false;
System.out.println(Thread.currentThread().getName() + "\t超过2s没有取到蛋糕，消费退出");
System.out.println();
return;
}
System.out.println(Thread.currentThread().getName() + "\t消费队列" + result + "成功");
}
}
public void stop() throws Exception {
flag = false;
}
}
运行结果：
到此这篇关于java之阻塞队列BlockingQueue解析的文章就介绍到这了,更多相关java阻塞队列BlockingQueue内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
EasyExcel是阿里巴巴开源的一个excel处理框架,以使用简单、节省内存著称,这篇文章主要为大家介绍了如何利用EasyExcel导出合并单元格,需要的可以参考下2023-09-09
在我们的工作中,常常要对一个字符串进行一些操作,这里提供一些常用的方法,常常需要这些方法进行组合处理字符串,这篇文章主要给大家介绍了关于JAVA String类中的一些常用方法,需要的朋友可以参考下2023-10-10
这篇文章主要介绍了java实现日期拆分的方法,基于java日期类实现对日期字符串的拆分功能,具有一定参考借鉴价值,需要的朋友可以参考下2015-07-07
并查集是一种用来管理元素分组情况的数据结构。并查集可以高效地进行如下操作。本文将通过Java实现并查集，感兴趣的小伙伴可以了解一下2022-01-01
springboot3整合knife4j详细图文教程(swagger增强)
开发api提供对应的接口规范进行联调或并行开发,api文档管理必不可少,常用的Knife4j基于swagger(依赖已经compile),可以进行管理,下面这篇文章主要给大家介绍了关于springboot3整合knife4j的相关资料,需要的朋友可以参考下2024-03-03
这篇文章主要介绍了springboot如何实现接口版本控制,接口版本控制,比如微服务请求中某个接口需要升级,正常做法是升级我们的版本,文中有详细的代码示例供大家参考,具有一定的参考价值,需要的朋友可以参考下2024-03-03
Java中的List接口从Java 8开始新增了stream()方法,用于创建一个Stream流对象,这篇文章主要给大家介绍了关于java中List常用的4种stream()方法的相关资料,需要的朋友可以参考下2024-02-02
这篇文章主要介绍了Java如何使用SSLContext请求https链接问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-01-01
这篇文章给大家介绍了三步实现SpringBoot全局日志记录,整合Echarts实现数据大屏,文中通过代码示例给大家介绍的非常详细,具有一定的参考价值,需要的朋友可以参考下2024-03-03
这篇文章主要由浅入深为大家介绍了springboot中使用redis的方法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2017-11-11
最新评论