20.并发容器之ArrayBlockingQueue和LinkedBlockingQueue实现原理详解
1. ArrayBlockingQueue简介
在多线程编程过程中,为了业务解耦和架构设计,经常会使用并发容器用于存储多线程间的共享数据,这样不仅可以保证线程安全,还可以简化各个线程操作。例如在“生产者-消费者”问题中,会使用阻塞队列(BlockingQueue)作为数据容器,关于BlockingQueue可以看这篇文章。为了加深对阻塞队列的理解,唯一的方式是对其实验原理进行理解,这篇文章就主要来看看ArrayBlockingQueue和LinkedBlockingQueue的实现原理。
2. ArrayBlockingQueue实现原理
阻塞队列最核心的功能是,能够可阻塞式的插入和删除队列元素。当前队列为空时,会阻塞消费数据的线程,直至队列非空时,通知被阻塞的线程;当队列满时,会阻塞插入数据的线程,直至队列未满时,通知插入数据的线程(生产者线程)。那么,多线程中消息通知机制最常用的是lock的condition机制,关于condition可以看这篇文章的详细介绍。那么ArrayBlockingQueue的实现是不是也会采用Condition的通知机制呢?下面来看看。
2.1 ArrayBlockingQueue的主要属性
ArrayBlockingQueue的主要属性如下:
/** The queued items */final Object[] items;/** items index for next take, poll, peek or remove */int takeIndex;/** items index for next put, offer, or add */int putIndex;/** Number of elements in the queue */int count;/* * Concurrency control uses the classic two-condition algorithm * found in any textbook. *//** Main lock guarding all access */final ReentrantLock lock;/** Condition for waiting takes */private final Condition notEmpty; //消费者等待线程/** Condition for waiting puts */private final Condition notFull; //生产者等待线程
从源码中可以看出ArrayBlockingQueue内部是采用数组进行数据存储的(属性items
),为了保证线程安全,采用的是ReentrantLock lock,为了保证可阻塞式的插入删除数据利用的是Condition,当获取数据的消费者线程被阻塞时会将该线程放置到notEmpty等待队列中,当插入数据的生产者线程被阻塞时,会将该线程放置到notFull等待队列中。而notEmpty和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和take方法是怎样实现的。
2.2 put方法详解
put(E e)
方法源码如下:
public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { //如果当前队列已满,将线程移入到notFull等待队列中 while (count == items.length) notFull.await(); //移入生产者等待线程 //满足插入数据的要求,直接进行入队操作 enqueue(e); } finally { lock.unlock(); }}
该方法的逻辑很简单,当队列已满时(count == items.length
)将线程移入到notFull等待队列中,如果当前满足插入数据的条件,就可以直接调用enqueue(e)
插入数据元素。enqueue方法源码为:
private void enqueue(E x) { // assert lock.getHoldCount() == 1; // assert items[putIndex] == null; final Object[] items = this.items; //插入数据 items[putIndex] = x; if (++putIndex == items.length) putIndex = 0; count++; //通知消费者线程,当前队列中有数据可供消费 notEmpty.signal();}
enqueue方法的逻辑同样也很简单,先完成插入数据,即往数组中添加数据(items[putIndex] = x
),然后通知被阻
塞的消费者线程,当前队列中有数据可供消费(notEmpty.signal()
)。
2.3 take方法详解
take方法源码如下:
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { //如果队列为空,没有数据,将消费者线程移入等待队列中 while (count == 0) notEmpty.await();//移入消费者等待线程 //获取数据 return dequeue(); } finally { lock.unlock(); }}
take方法也主要做了两步:1. 如果当前队列为空的话,则将获取数据的消费者线程移入到等待队列中;2. 若队列不为空则获取数据,即完成出队操作dequeue
。dequeue方法源码为:
private E dequeue() { // assert lock.getHoldCount() == 1; // assert items[takeIndex] != null; 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.signal(); return x;}
dequeue方法也主要做了两件事情:1. 获取队列中的数据,即获取数组中的数据元素((E) items[takeIndex]
);2. 通知notFull等待队列中的线程,使其由等待队列移入到同步队列中,使其能够有机会获得lock,并执行完成功退出。
从以上分析,可以看出put和take方法主要是通过condition的通知机制来完成可阻塞式的插入数据和获取数据。在理解ArrayBlockingQueue后再去理解LinkedBlockingQueue就很容易了。
3. LinkedBlockingQueue实现原理
LinkedBlockingQueue是用链表实现的有界阻塞队列,当构造对象时为指定队列大小时,队列默认大小为Integer.MAX_VALUE
。从它的构造方法可以看出:
public LinkedBlockingQueue() { this(Integer.MAX_VALUE);}
3.1 LinkedBlockingQueue的主要属性
LinkedBlockingQueue的主要属性有:
/** Current number of elements */private final AtomicInteger count = new AtomicInteger();/** * Head of linked list. * Invariant: head.item == null */transient Node<E> head;//头/** * Tail of linked list. * Invariant: last.next == null */private transient Node<E> last; //尾/** Lock held by take, poll, etc */private final ReentrantLock takeLock = new ReentrantLock(); //消费者可重入锁/** Wait queue for waiting takes */private final Condition notEmpty = takeLock.newCondition();//消费者线程/** Lock held by put, offer, etc */private final ReentrantLock putLock = new ReentrantLock(); //生产者可重入锁/** Wait queue for waiting puts */private final Condition notFull = putLock.newCondition();//生产者线程
可以看出与ArrayBlockingQueue主要的区别是,LinkedBlockingQueue在插入数据和删除数据时分别是由两个不同的lock(takeLock
和putLock
)来控制线程安全的,因此,也由这两个lock生成了两个对应的condition(notEmpty
和notFull
)来实现可阻塞的插入和删除数据。并且,采用了链表的数据结构来实现队列,这样设计的目的是因为插入数据
是往队尾进行的,删除数据是往队头进行的,从而可以降低获取不到Lock锁而进入Waiting状态,提高并发
Node结点的定义为:
static class Node<E> {E item;/*** One of:* - the real successor Node* - this Node, meaning the successor is head.next* - null, meaning there is no successor (this is the last node)*/Node<E> next;Node(E x) { item = x; } }
接下来,我们也同样来看看put方法和take方法的实现。
3.2 put方法详解
put方法源码为:
public void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();// Note: convention in all put/take/etc is to preset local var// holding count negative to indicate failure unless set.int c = -1;Node<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;putLock.lockInterruptibly();//与lock方法类似,优先考虑响应中断try {/** Note that count is used in wait guard even though it is* not protected by lock. This works because count can* only decrease at this point (all other puts are shut* out by lock), and we (or some other waiting put) are* signalled if it ever changes from capacity. Similarly* for all other uses of count in other wait guards.*///如果队列已满,则阻塞当前线程,将其移入等待队列while (count.get() == capacity) {notFull.await(); //生产者线程阻塞}//入队操作,向队尾插入数据enqueue(node);c = count.getAndIncrement();//索引后移//若队列满足插入数据的条件,则通知被阻塞的生产者线程if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}if (c == 0)signalNotEmpty(); }
put方法的逻辑也同样很容易理解,可见注释。基本上和ArrayBlockingQueue的put方法一样。take方法的源码如下:
public E take() throws InterruptedException {E x;int c = -1;final AtomicInteger count = this.count;final ReentrantLock takeLock = this.takeLock;takeLock.lockInterruptibly();try {//当前队列为空,则阻塞当前线程,将其移入到等待队列中,直至满足条件while (count.get() == 0) {notEmpty.await();}//移除队头元素,获取数据x = dequeue();c = count.getAndDecrement();//如果当前满足移除元素的条件,则通知被阻塞的消费者线程if (c > 1)notEmpty.signal();} finally {takeLock.unlock();}if (c == capacity)signalNotFull();return x; }
take方法的主要逻辑请见于注释,也很容易理解。
4. ArrayBlockingQueue与LinkedBlockingQueue的比较
相同点:ArrayBlockingQueue和LinkedBlockingQueue都是通过condition通知机制来实现可阻塞式插入和删除元素,并满足线程安全的特性;
不同点:1. ArrayBlockingQueue底层是采用的数组进行实现,而LinkedBlockingQueue则是采用链表数据结构;
ArrayBlockingQueue插入和删除数据,只采用了一个lock,而LinkedBlockingQueue则是在插入和删除分别采用了
putLock
和`takeLock`,这样可以降低线程由于线程无法获取到lock而进入WAITING状态的可能性,从而提高了线程并发执行的效率。
转载于:https://www.cnblogs.com/itxiaok/p/10356593.html
20.并发容器之ArrayBlockingQueue和LinkedBlockingQueue实现原理详解相关推荐
- Java 并发编程CAS、volatile、synchronized原理详解
CAS(CompareAndSwap) 什么是CAS? 在Java中调用的是Unsafe的如下方法来CAS修改对象int属性的值(借助C来调用CPU底层指令实现的): /*** * @param o ...
- 【转】Java并发编程:并发容器之ConcurrentHashMap
JDK5中添加了新的concurrent包,相对同步容器而言,并发容器通过一些机制改进了并发性能.因为同步容器将所有对容器状态的访问都串行化了,这样保证了线程的安全性,所以这种方法的代价就是严重降低了 ...
- Java并发编程:并发容器之CopyOnWriteArrayList(转载)
Java并发编程:并发容器之CopyOnWriteArrayList(转载) 原文链接: http://ifeve.com/java-copy-on-write/ Copy-On-Write简称COW ...
- Java阻塞队列ArrayBlockingQueue和LinkedBlockingQueue实现原理分析
转载自 Java阻塞队列ArrayBlockingQueue和LinkedBlockingQueue实现原理分析 Java中的阻塞队列接口BlockingQueue继承自Queue接口. Block ...
- 并发编程五:java并发线程池底层原理详解和源码分析
文章目录 java并发线程池底层原理详解和源码分析 线程和线程池性能对比 Executors创建的三种线程池分析 自定义线程池分析 线程池源码分析 继承关系 ThreadPoolExecutor源码分 ...
- 【面试:并发篇09:多线程:interrupt 方法详解】
[面试:并发篇09:多线程:interrupt 方法详解] 00.前言 如果有任何问题请指出,感谢. 01.介绍 程序中,有些线程的中断需要外部干预,比如线程中存在while(true)循环,或者存在 ...
- python小游戏代码大全-20行python代码的入门级小游戏的详解
背景: 作为一个python小白,今天从菜鸟教程上看了一些python的教程,看到了python的一些语法,对比起来(有其他语言功底),感觉还是非常有趣,就随手添了一点内容,改了一个小例程,当着练练手 ...
- Linux进程最大socket数,Linux下高并发socket最大连接数所受的各种限制(详解)
1.修改用户进程可打开文件数限制 在Linux平台上,无论编写客户端程序还是服务端程序,在进行高并发TCP连接处理时,最高的并发数量都要受到系统对用户单一进程同时可打开文件数量的限制(这是因为系统为每 ...
- 碧蓝航线8.20服务器维护,碧蓝航线8月20日更新内容及新玩法活动详解介绍
碧蓝航线8月20日更新了什么内容?游戏在今天上线了限时活动,还有全新的角色哦!很多小伙伴们可能还不知道具体有哪些活动和玩法吧!下面是小编带来的攻略解析,感兴趣的可以一起来关注下哦! 碧蓝航线8月20日 ...
- 并发容器之ThreadLocal详解
文章目录 ThreadLocal的简介 ThreadLocal的实现原理 ThreadLocalMap详解 Entry数据结构 set方法 getEntry方法 remove方法 ThreadLoca ...
最新文章
- GPU、AI芯片技术市场分析
- 干货 | OpenCV看这篇就够了,9段代码详解图像变换基本操作
- 基于Java Swing的仪表盘实现
- OpenCV单kinect多帧静止场景的深度图像去噪
- 对一致性Hash算法,Java代码实现的深入研究
- loadRUnner常用函数
- tensorflow随笔-变量
- Lucene概述第一部分:创建索引
- 辉哥给rockchip修复了一个内存溢出问题
- saslauthd mysql_Postfix,saslauthd,mysql,smtp身份验证问题
- .net开发MIS系统的难处.
- windows中的服务隔离 service isolation and service SID(Virtual Account)
- 天堂2单机版如何架设mysql_[JAVA版本]新人入门级单机游戏架设教程
- 《统计学》第八版贾俊平第一章课后习题及答案总结
- NS3中路由协议分析【AODV理论分析】
- 华中数控系统与高创驱动器EtherCat通讯设置及控制参数
- 2021年中国国内生产总值(GDP)、GDP结构及人均国内生产总值分析[图]
- java毕业生设计高考填报信息系统计算机源码+系统+mysql+调试部署+lw
- LASSO和LARS
- 点击导航栏切换不同的页面