java 多线程阻塞队列 与 阻塞方法与和非阻塞方法
Queue是什么
队列,是一种数据结构。除了优先级队列和LIFO队列外,队列都是以FIFO(先进先出)的方式对各个元素进行排序的。无论使用哪种排序方式,队列的头都是调用remove()或poll()移除元素的。在FIFO队列中,所有新元素都插入队列的末尾。队列都是线程安全的,内部已经实现安全措施,不用我们担心
Queue中的方法
Queue中的方法不难理解,6个,每2对是一个也就是总共3对。看一下JDK API就知道了:
注意一点就好,Queue通常不允许插入Null,尽管某些实现(比如LinkedList)是允许的,但是也不建议。
ArrayBlockingQueue:基于数组实现的一个阻塞队列,在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性,默认情况下为非公平的,即不保证等待时间最长的队列最优先能够访问队列。
LinkedBlockingQueue:基于链表实现的一个阻塞队列,在创建LinkedBlockingQueue对象时如果不指定容量大小,则默认大小为Integer.MAX_VALUE。
PriorityBlockingQueue:以上2种队列都是先进先出队列,而PriorityBlockingQueue却不是,它会按照元素的优先级对元素进行排序,按照优先级顺序出队,每次出队的元素都是优先级最高的元素。注意,此阻塞队列为无界阻塞队列,即容量没有上限(通过源码就可以知道,它没有容器满的信号标志),前面2种都是有界队列。
DelayQueue:基于PriorityQueue,一种延时阻塞队列,DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue也是一个无界队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。
注意:
1、必须要使用take()方法在获取的时候达成阻塞结果
2、使用poll()方法将产生非阻塞效果
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -817911632652898426L;
/** The queued items */
private final E[] items;
/** items index for next take, poll or remove */
private int takeIndex;
/** items index for next put, offer, or add. */
private int putIndex;
/** Number of items in the queue */
private int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
private final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
}
public ArrayBlockingQueue(int capacity) {
}
public ArrayBlockingQueue(int capacity, boolean fair) {
}
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
}
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
final E[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == items.length)
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
insert(e);
} finally {
lock.unlock();
}
}
private void insert(E x) {
items[putIndex] = x;
putIndex = inc(putIndex);
++count;
notEmpty.signal();
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == 0)
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
E x = extract();
return x;
} finally {
lock.unlock();
}
}
private E extract() {
final E[] items = this.items;
E x = items[takeIndex];
items[takeIndex] = null;
takeIndex = inc(takeIndex);
--count;
notFull.signal();
return x;
}
public class Test {
private int queueSize = 10;
private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
public static void main(String[] args) {
Test test = new Test();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
producer.start();
consumer.start();
}
class Consumer extends Thread{
@Override
public void run() {
consume();
}
private void consume() {
while(true){
synchronized (queue) {
while(queue.size() == 0){
try {
System.out.println("队列空,等待数据");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
queue.notify();
}
}
queue.poll(); //每次移走队首元素
queue.notify();
System.out.println("从队列取走一个元素,队列剩余"+queue.size()+"个元素");
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
produce();
}
private void produce() {
while(true){
synchronized (queue) {
while(queue.size() == queueSize){
try {
System.out.println("队列满,等待有空余空间");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
queue.notify();
}
}
queue.offer(1); //每次插入一个元素
queue.notify();
System.out.println("向队列取中插入一个元素,队列剩余空间:"+(queueSize-queue.size()));
}
}
}
}
}
public class Test {
private int queueSize = 10;
private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);
public static void main(String[] args) {
Test test = new Test();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
producer.start();
consumer.start();
}
class Consumer extends Thread{
@Override
public void run() {
consume();
}
private void consume() {
while(true){
try {
queue.take();
System.out.println("从队列取走一个元素,队列剩余"+queue.size()+"个元素");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
produce();
}
private void produce() {
while(true){
try {
queue.put(1);
System.out.println("向队列取中插入一个元素,队列剩余空间:"+(queueSize-queue.size()));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
转载于:https://www.cnblogs.com/signheart/p/6606475.html
java 多线程阻塞队列 与 阻塞方法与和非阻塞方法相关推荐
- JAVA物联所需技术_基于JAVA多线程技术解决物联云端服务雪崩效应的方法与流程...
本发明涉及互联网技术领域,特别涉及一种基于JAVA多线程技术解决物联云端服务雪崩效应的方法. 背景技术: 目前,物联云系统已经作为普遍的智能电视平台出现在我们面前,而细致分析物联云系统我们可以发现,当 ...
- linux c fifo阻塞写和非阻塞写,linux—FIFO的使用与非阻塞标志(O_NONBLOCK)的影响
FIFO文 #include #include int mkfifo( const char*pathname, mode_t mode); 参数: pathname:FIFO的路径名+文件名. mo ...
- 多线程-并发编程(7)-生产者消费者模式及非阻塞队列与阻塞队列实现
生产者消费者模式是一个十分经典的多线程协作模式 弄懂生产者消费者问题能够让我们对多线程编程的理解更加深刻 存在3个元素 1.生产者(类比厨师) 2.生产者的生产产品(类比美食) 3.消费者(类比吃货) ...
- java并发性是指什么_java – 什么是“非阻塞”并发,它与普通并发性有什么不同?...
What is Non-blocking Concurrency and how is it different. 正式: In computer science, non-blocking sync ...
- 非阻塞式编程 php,简单介绍PHP非阻塞模式
非阻塞模式是指利用socket事件的消息机制,Server端与Client端之间的通信处于异步状态. 让PHP不再阻塞当PHP作为后端处理需要完成一些长时间处理,为了快速响应页面请求,不作结果返回判断 ...
- php 非阻塞post请求,PHP实现的CURL非阻塞调用类
本文实例讲述了PHP实现的CURL非阻塞调用类.分享给大家供大家参考,具体如下: 前面一篇<PHP实现非阻塞模式的方法>文章讲述了PHP中实现非阻塞模式,其实如果只是HTTP的话,直接用C ...
- mysql异步非阻塞方式_如何理解swoole异步非阻塞?
传统的apache2handler或php-fpm本质上都是短生命周期(请求后释放资源)的FastCGI运行模式. 请求来了,master进程会调用worker进程来处理,处理完后释放资源. 假设你在 ...
- 乾坤大挪移,如何将同步阻塞(sync)三方库包转换为异步非阻塞(async)模式?Python3.10实现。
众所周知,异步并发编程可以帮助程序更好地处理阻塞操作,比如网络 IO 操作或文件 IO 操作,避免因等待这些操作完成而导致程序卡住的情况.云存储文件传输场景正好包含网络 IO 操作和文件 IO 操作, ...
- linux 管道非阻塞,在Linux中管道上的非阻塞读取
可以在管道上进行非阻塞I / O吗? fcntl无法设置O_NONBLOCK. Linux编程接口的页面918包括一个表'从管道读取n个字节或FIFO(p)'的语义.此表列出了管道和FIFO的行为,其 ...
最新文章
- Docker中的Java内存消耗优化以及我们如何使用Spring Boot
- Linux环境编译qtmqtt,qtmqttclient
- PHP JSON_ENCODE 不转义中文汉字的方法
- 如何在SQL Server 2005中使用作业实现备份和特定删除
- Android图片360全景旋转
- php中的boolean(布尔)类型
- spring异常处理实例(登录例子)
- 有哪些开源C ++静态分析工具? [关闭]
- 输入输出系统的发展概况
- 安卓第十三天笔记-服务(Service)
- 京东发布《未来科技趋势白皮书》,101页详解5大关键技术(附PDF下载)
- VMware-ESXi-6.7.0许可证
- 第二课:更换国内下载源(阿里源为例)
- 新浪邮箱服务器密码被盗,邮箱密码盗取严重的注意事项以及处理建议
- 图书馆管理系统(PHP期末报告)
- 域用户的管理之一次同时对多个用户进行管理
- 侯捷c++课程学习一
- amazon mechanical turk介绍
- JS 正则表达式否定匹配(正向前瞻)
- H5项目中 ios的border不显示问题
热门文章
- r语言赋值为na_r语言将空白格替换成NA
- java内存管理课程设计_Java内存管理分析
- 查看PLC IP 端口_西门子828D数控系统X130接口通讯怪异现象(X130手动设置的 IP)...
- 排除服务器简单系统故障方法,引导CD排除服务器故障方法有哪些?
- mysql生产环境加索引_【生产篇】_MySQL环境下如何查看基于表的索引定义
- JavaScript学习随记——常见全局对象属性及方法
- css设置a连接禁用样式_使用CSS禁用链接
- python 示例_带有示例的Python字典popitem()方法
- 使用方法实现数组的对调与输出
- python---文件处理