潍坊企业网站价格,怎样在手机上做自己的网站,163邮箱企业邮箱,app网站建站系统策划方案文章目录 Pre生产者-消费者模式阻塞队列 vs 普通队列JUC提供的7种适合与不同应用场景的阻塞队列插入操作#xff1a;添加元素到队列中移除操作#xff1a;从队列中移除元素。 ArrayBlockingQueue源码解析类结构指定初始容量及公平/非公平策略的构造函数根据已有集合初始化队列… 文章目录 Pre生产者-消费者模式阻塞队列 vs 普通队列JUC提供的7种适合与不同应用场景的阻塞队列插入操作添加元素到队列中移除操作从队列中移除元素。 ArrayBlockingQueue源码解析类结构指定初始容量及公平/非公平策略的构造函数根据已有集合初始化队列的构造函数ArrayBlockingQueue主要属性出队和入队的环形队列插入元素1. offer(E e)方法非阻塞插入元素2. addE e方法非阻塞插入元素3. putE e方法阻塞式插入元素4. offeretimeunit阻塞式插入元素 移除元素1. poll()方法非阻塞移除元素2. remove()方法非阻塞移除元素3. take()方法阻塞式移除元素4.poll(long timeout, TimeUnit unit)阻塞式移除元素 小结 LinkedBlockingQueue源码解析LinkedBlockingQueue类结构LinkedBlockingQueue的主要属性LinkedBlockingQueue的三个构造函数插入元素移除元素小结 ArrayBlockingQueue vs LinkedBlockingQueue 小结 Pre
J.U.C Review - 阻塞队列原理/源码分析
生产者-消费者模式
阻塞队列的使用场景一般是在“生产者-消费者”模式中在该模式中“生产者”和“消费者”相互独立两者之间的通信依靠阻塞队列完成。
生产者将待处理的数据放入“队列”中随后消费者从该“队列”中取出数据进行后续处理。“生产者-消费者”模式简化了开发流程消除了“生产者”和“消费者”对代码实现的依赖生产者只需将生成的数据放入队列中而不需要知道由谁消费以及何时和如何消费同样消费者也不需要知道数据是谁生产的这样就实现了将“生产者”和“消费者”两者操作解耦。
在JUC中线程池本质上就是一个“生产者-消费者”模式的实现提交任务的线程是生产者线程池中线程是消费者在线程池中当提交的任务不能被立即执行的时候线程池就会将提交的任务放到一个阻塞的任务队列中。我们可以根据业务场景的需求选用相应类型的阻塞队列来构造不同功能的线程池。 阻塞队列 vs 普通队列
阻塞队列与普通队列的区别在于阻塞队列提供了可阻塞的put和take方法。
如果队列是空的消费者使用take方法从队列中获取数据就会被阻塞直到队列有数据可用当队列是满的生产者使用put方法向队列里添加数据就会被阻塞直到队列中数据被消费有空闲位置可用 JUC提供的7种适合与不同应用场景的阻塞队列
JUC提供了7种适合与不同应用场景的阻塞队列。
1ArrayBlockingQueue基于数组实现的有界阻塞队列。2LinkedBlockingQueue基于链表实现的有界阻塞队列。3PriorityBlockingQueue支持按优先级排序的无界阻塞队列。4DelayQueue优先级队列实现的无界阻塞队列。5SynchronousQueue不存储元素的阻塞队列。6LinkedTransferQueue基于链表实现的无界阻塞队7LinkedBlockingDeque基于链表实现的双向无界阻塞队列。
7个阻塞队列全部实现了BlockingQueue接口插入和移除元素分别各提供了4种处理方式。
插入操作添加元素到队列中
1adde当队列满的时候继续插入元素会抛出IllegalStateExceptionQueue full异常。2offere当队列满的时候继续插入元素不会阻塞直接返回false。如果插入成功则返回true。3pute当队列满的时候继续插入元素线程会被一直阻塞直到队列有空闲位置可用时为止。4offeretimeunit当队列满的时候调用该方法的线程会被阻塞一段时间如果超过指定时间还未添加成功线程直接退出。 移除操作从队列中移除元素。
1remove()当队列为空时调用该方法元素会抛出NoSuchElementException异常。2poll()当队列为空时调用该方法不会阻塞直接返回null。当队列不为空时则从队列中取出一个元素。3take()当队列为空时调用该方法会被一直阻塞直到队列有数据可用时为止4polltimeunit当队列为空时调用该方法的线程会被阻塞一段时间如果超过指定时间队列仍没有数据可用直接返回null。 当然如果是向无界阻塞队列中插入元素因为无界队列永远不可能会出现满的情况所以使用put或offerretimeunit方法永远不会被阻塞 ArrayBlockingQueue源码解析
Java Review - 并发编程_ArrayBlockingQueue原理源码剖析
ArrayBlockingQueue是一个基于数组实现的有界阻塞队列。此队列按照先进先出FIFO的原则对元素进行排序。
类结构 指定初始容量及公平/非公平策略的构造函数 /*** 创建一个具有给定固定容量和默认访问策略的 {code ArrayBlockingQueue}。** param capacity 该队列的容量* throws IllegalArgumentException 如果 {code capacity 1}*/public ArrayBlockingQueue(int capacity) {this(capacity, false); // 使用默认的非公平访问策略}/*** 创建一个具有给定固定容量和指定访问策略的 {code ArrayBlockingQueue}。** param capacity 该队列的容量* param fair 如果 {code true}则被阻塞在插入或移除操作上的线程按 FIFO 顺序处理* 如果 {code false}则访问顺序是未指定的。* throws IllegalArgumentException 如果 {code capacity 1}*/public ArrayBlockingQueue(int capacity, boolean fair) {if (capacity 0)throw new IllegalArgumentException(); // 容量必须大于 0this.items new Object[capacity]; // 初始化存储数组lock new ReentrantLock(fair); // 创建 ReentrantLock根据 fair 参数决定是否使用公平锁notEmpty lock.newCondition(); // 创建一个表示队列非空的条件notFull lock.newCondition(); // 创建一个表示队列非满的条件}
构造函数实现代码如上所示它属于核心构造函数按照指定的容量初始化items数组队列的容量在构造时指定一旦指定就不能修改.
该函数支持公平/非公平策略在构造器中以指定的策略实例化lock全局锁使用ReentrantLock独占锁保证同时只能有一个线程进行入队和出队的操作。
fairtrue为公平策略表示所有线程严格按照请求的顺序添加或删除元素fairtrue为非公平策略表示允许新申请线程“插队”即新申请线程和被唤醒的线程谁先抢占到锁谁就能往队列中添加/删除元素。此外还创建了两个Condition条件队列对象。当队列满时申请插入元素的线程需要在notFull上等待当队列空时申请获取元素的线程会在notEmpty上等待 根据已有集合初始化队列的构造函数 /*** 创建一个具有给定固定容量、指定访问策略并且初始包含给定集合中元素的 {code ArrayBlockingQueue}* 元素按照集合迭代器的遍历顺序添加。** param capacity 该队列的容量* param fair 如果 {code true}则被阻塞在插入或移除操作上的线程按 FIFO 顺序处理* 如果 {code false}则访问顺序是未指定的。* param c 初始包含的元素集合* throws IllegalArgumentException 如果 {code capacity} 小于 {code c.size()} 或小于 1。* throws NullPointerException 如果指定的集合或其任何元素为 null。*/public ArrayBlockingQueue(int capacity, boolean fair,Collection? extends E c) {this(capacity, fair); // 调用构造函数初始化队列final ReentrantLock lock this.lock;lock.lock(); // 仅为了可见性锁定而不是互斥try {final Object[] items this.items;int i 0;try {for (E e : c)items[i] Objects.requireNonNull(e); // 将集合中的元素添加到队列中} catch (ArrayIndexOutOfBoundsException ex) {throw new IllegalArgumentException(); // 如果索引超出范围抛出 IllegalArgumentException}count i; // 更新队列中的元素数量putIndex (i capacity) ? 0 : i; // 更新下一个插入位置的索引} finally {lock.unlock(); // 解锁}}
集合c的长度不能大于capacity否则会抛出IllegalArgumentException异常。以集合c初始化队列时使用了lock锁这是为了保证数据的可见性保证在构造器结束后items、count和putIndex的修改对所有线程可见.
ArrayBlockingQueue在构造时指定容量后面不能被改变。ArrayBlockingQueue是基于数组构建的它的内部有一个数组items用于存储队列元素 ArrayBlockingQueue主要属性 /** 队列中的元素 */final Object[] items;/** 下一次 take、poll、peek 或 remove 操作的 items 索引 */int takeIndex;/** 下一次 put、offer 或 add 操作的 items 索引 */int putIndex;/** 队列中的元素数量 */int count;/** 主锁保护所有访问 */final ReentrantLock lock;/** 等待 take 操作的条件 */private final Condition notEmpty;/** 等待 put 操作的条件 */private final Condition notFull; 1putIndex字段初始值0表示下一个入队元素的数组索引takeIndex为队头位置。 2takeIndex字段初始值0表示下一个出队元素的数组索引putIndex为队尾位置。 3count字段同于存储队列中当前元素个数。 4lock字段全局ReentrantLock锁。使用ReentrantLock锁保证元素入队和出队操作的线程安全保证同时只能有一个线程进行出队入队的操作。我们知道ReentrantLock锁有公平和非公平两种实现可以在ArrayBlockingQueue构造时指定lock锁的类型。 5notEmpty字段是lock的一个Condition实例。当队列为空时获取元素的线程会被阻塞并在该队列等待被唤醒。当队列有数据可用时其他线程会调用notEmpty.signal()唤醒等待中线程。 6notFull字段是lock的一个Condition实例。当队列已满时插入元素的线程会在该队列等待被唤醒。当队列有空闲位置时其他线程会调用notFull. signal()唤醒等待中线程。 出队和入队的环形队列
我们再看插入元素和移除元素方法源码前需要先对ArrayBlockingQueue的出队和入队流程有个总体的认识这样后面理解源码时就会更容易。
ArrayBlockingQueue在进行不断的元素入队和出队操作时内部数组items、putIndex、takeIndex三者其实构成了一个环形结构 当ArrayBlockingQueue被初始化时takeIndex和putIndex都指向item索引0的位置假设队列的容量为n。 当生产者线程向队列中插入元素时如果队列有空闲位置元素会被插入items[putIndex]中putIndex会向后移动。 当putIndex移动到数组items最后一个索引位置n-1时如果这时继续向队列中插入元素元素会被插入到items[n-1]putIndex会重新指向item第一个索引位置0。putIndex的移动过程其实就是一个环形循环的过程 当消费者线程从队列中获取元素时如果队列是非空状态队列将item [takeIndex]位置的元素返回后takeIndex会向后移动。takeIndex的移动过程和putIndex一样都是环形循环的过程。 假设阻塞队列的容量为6。item数组容量为6 1初始状态 : 元素个数count0put索引putIndex0take索引takeIndex0 . 如果此时使用take方法移除元素take线程会被阻塞。
2插入元素1 . 线程调用put方法插入元素1putIndex向后移动putIndex。put操作结束后count1putIndex1takeIndex0
3继续插入元素2至元素5. put操作结束后count5putIndex5takeIndex0
4继续插入元素6. putIndex向后移putIndex6超过数组最大索引重置为0。put操作结束后count6putIndex0takeIndex0。此时队列是满的如果继续插入元素put线程会被阻塞 5移除元素. 线程调用take方法移除元素从队列中取出元素1。takeIndex向后移takeIndex1。take结束后count5putIndex0takeIndex1 6继续移除元素2至元素5. take操作结束后count1putIndex0takeIndex5 7继续移除元素. takeIndex向后移takeIndex6超过数组最大索引重置为0。take操作结束后count0putIndex0takeIndex0 . 这时队列是空的如果继续移除元素take线程会被塞。 插入元素
插入元素的逻辑很简单用同一个ReentrantLock独占锁lock实例保证同时只有一个线程执行插入操作如果队列中有空闲位置元素插入队尾。否则队列是满的ArrayBlock-ingQueue提供了4种不同的处理方式offerE eaddE eputE e、offeretimeu-nit。其中前两个插入元素时不会阻塞线程后两个方法在队列满时会阻塞线程。
1. offer(E e)方法非阻塞插入元素
线程使用该方法向插入元素时如果队列中有空闲位置插入成功并返回true。如果队列是满的线程也不会被阻塞而是直接返回false。
/*** 将指定的元素插入到此队列的尾部如果可以立即完成而不会超过队列的容量则返回 {code true}* 否则如果队列已满则返回 {code false}。此方法通常优于 {link #add} 方法* 因为后者在无法插入元素时只会抛出异常。** param e 要插入的元素* return 如果元素成功插入则返回 {code true}否则返回 {code false}* throws NullPointerException 如果指定的元素为 null*/
public boolean offer(E e) {// 确保插入的元素不为 nullObjects.requireNonNull(e);// 获取锁以确保插入过程中的线程安全final ReentrantLock lock this.lock;lock.lock();try {// 检查队列是否已满if (count items.length)return false;else {// 将元素插入队列的尾部enqueue(e);return true;}} finally {// 释放锁以避免死锁lock.unlock();}
}
1首先使用lock. lock()加锁保证只能有一个线程执行入队操作。2如果队列是满的插入元素失败直接返回false。3如果队列未满调用私有方法enqueuee插入元素到队尾插入成功后返回true。4最后操作完毕后释放锁。
enqueue插入元素的核心方法插入元素到队尾。
/*** 在当前的 put 位置插入元素前进位置并发出信号。* 只有在持有锁的情况下才能调用此方法。*/
private void enqueue(E e) {// 断言当前线程持有锁// assert lock.isHeldByCurrentThread();// 断言锁的持有计数为 1// assert lock.getHoldCount() 1;// 断言当前 put 位置为空// assert items[putIndex] null;final Object[] items this.items;items[putIndex] e; // 将元素存储到当前的 put 位置if (putIndex items.length) putIndex 0; // putIndex 增加 1如果达到数组长度则重置为 0count; // 队列中的元素数量增加 1notEmpty.signal(); // 唤醒在 notEmpty 条件上等待的线程
}
主要步骤如下
1首先存储插入元素到items[putIndex]。2对尾位置putIndex向后移1位如果putIndex达到最大索引重新定位到索引0。3队列中元素个数加1。4唤醒notEmpt上一个等待线程该线程执行后续移除元素操作。 2. addE e方法非阻塞插入元素
线程使用该方法向插入元素时如果队列中有空闲位置插入成功并返回true。如果队列是满的线程也不会被阻塞而是直接抛出异常。
add方法比较简单直接调用了父类AbstractQueue的add方法。一般这种调用方式采用模板模式的写法可以使用模板方法来解决通用流程的实现
/*** 将指定的元素插入到此队列的尾部如果可以立即完成而不会超过队列的容量则返回 {code true}* 否则如果队列已满则抛出 {code IllegalStateException} 异常。** param e 要添加的元素* return {code true}符合 {link Collection#add} 的规定* throws IllegalStateException 如果队列已满* throws NullPointerException 如果指定的元素为 null*/
public boolean add(E e) {return super.add(e); // 调用父类的 add 方法进行元素添加
}
AbstractQueue. add方法很简单首先调用ArrayBlockingQueue的offere插入元素如果插入成功直接返回ture否则抛出IllegalStateException(“Queue full”)异常
/*** 将指定的元素插入到此队列的尾部如果可以立即完成而不会超过队列的容量则返回 {code true}* 否则如果队列已满则抛出 {code IllegalStateException} 异常。** param e 要添加的元素* return {code true}符合 {link Collection#add} 的规定* throws IllegalStateException 如果队列已满* throws NullPointerException 如果指定的元素为 null*/
public boolean add(E e) {if (offer(e)) { // 尝试使用 offer 方法插入元素return true; // 插入成功返回 true} else {throw new IllegalStateException(Queue full); // 队列已满抛出异常}
} 3. putE e方法阻塞式插入元素
线程使用该方法向插入元素时如果队列是满的线程会被一直阻塞并进入notFull条件队列等待。否则就调用enqueue方法插入元素
/*** 将指定的元素插入到此队列的尾部如果队列已满则等待空间变得可用。** param e 要插入的元素* throws InterruptedException 如果当前线程被中断* throws NullPointerException 如果指定的元素为 null*/
public void put(E e) throws InterruptedException {Objects.requireNonNull(e); // 确保插入的元素不为 nullfinal ReentrantLock lock this.lock;lock.lockInterruptibly(); // 使用可中断的锁获取机制try {while (count items.length) { // 如果队列已满notFull.await(); // 等待 notFull 条件信号直到队列中有空闲空间}enqueue(e); // 插入元素到队列尾部} finally {lock.unlock(); // 释放锁}
} 1判断插入元素不能为Null否则抛出NullPointerException异常。 2使用lock. lockInterruptibly()加锁保证只能有一个线程执行入队操作。 3如果队列是满的当前线程会被阻塞让出lock锁并在notFull条件队列中等待被其他线程唤醒。 线程被唤醒后需重新尝试获取锁获取锁成功后再次判断队列是否有空闲位置可用如果队列是满的线程再次被阻塞。 final ReentrantLock lock this.lock;入队和出队时使用了同一把独占锁也就是说同时只有一个线程进行出队和入队操作 线程被唤醒后为什么还需要再次判断 ? 其实这里使用while循环有两个原因 其一防止虚假唤醒防止线程被意外唤醒不经再次判断就直接调用enqueue方法。其二如果ArrayBlockingQueue使用的非公平锁策略则在本线程被其他线程唤醒后有可能新线程抢先完成了入队操作 4如果队列有空闲位置可用调用私有方法enqueuee插入元素 5最后操作完毕后释放锁。 4. offeretimeunit阻塞式插入元素
线程使用该方法向插入元素时如果队列是满的线程会被阻塞一段时间如果超过指定时间还未添加成功线程直接退出返回false。如果插入元素成功返回true
/*** 将指定的元素插入到此队列的尾部如果队列已满则最多等待指定的时间以等待空间变得可用。** param e 要插入的元素* param timeout 最大等待时间* param unit 时间单位* return 如果在指定时间内成功插入元素则返回 {code true}否则返回 {code false}* throws InterruptedException 如果当前线程被中断* throws NullPointerException 如果指定的元素为 null*/
public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {Objects.requireNonNull(e); // 确保插入的元素不为 nulllong nanos unit.toNanos(timeout); // 将指定的时间转换为纳秒final ReentrantLock lock this.lock;lock.lockInterruptibly(); // 使用可中断的锁获取机制try {while (count items.length) { // 如果队列已满if (nanos 0L) { // 如果剩余等待时间小于等于 0return false; // 返回 false 表示插入失败}nanos notFull.awaitNanos(nanos); // 等待 notFull 条件信号直到队列中有空闲空间或超时}enqueue(e); // 插入元素到队列尾部return true; // 返回 true 表示插入成功} finally {lock.unlock(); // 释放锁}
} 1判断插入元素不能为Null否则抛出NullPointerException异常。 2使用lock. lockInterruptibly()加锁保证只能有一个线程执行入队操作。 3如果队列是满的当前线程会被阻塞让出lock锁并在notFull条件队列中等待被其他线程唤醒。线程被唤醒后需重新尝试获取锁。 线程在两种情况下会被唤醒。 情况1等待超时被唤醒并获取到锁。线程会重新判断是否有空闲位置如果队列仍然是满的直接返回false。情况2被其他线程唤醒并获取到锁。线程被唤醒后重新判断是否有空闲位置如果线程仍然是满的并且等待超时直接返回false。 4如果队列有空闲位置可用调用私有方法enqueuee插入元素。 5最后操作完毕后释放锁。 移除元素
移除元素的逻辑也很简单使用同一个ReentrantLock独占锁lock保证同时只有一个线程执行移除操作如果队列中不为空从队头取出一个元素。如果队列是空的Array-BlockingQueue也提供了4种不同的处理方式remove()、poll()、take()、polltimeunit。
其中前两个移除元素时不会阻塞线程后两个方法在队列为空时会阻塞线程 1. poll()方法非阻塞移除元素
线程使用该方法移除元素时如果队列不为空从队头移除一个元素并返回。如果队列是空的线程不会被阻塞而是直接返回null。
/*** 从队列的头部移除并返回元素如果队列为空则返回 {code null}。** return 如果队列不为空则返回队列头部的元素否则返回 {code null}*/
public E poll() {final ReentrantLock lock this.lock;lock.lock(); // 获取锁以确保线程安全try {if (count 0) { // 如果队列为空return null; // 返回 null} else {return dequeue(); // 移除并返回队列头部的元素}} finally {lock.unlock(); // 释放锁}
} 首先使用lock. lock()加锁保证只能有一个线程执行入队操作。 如果队列是空的直接返回null。 如果队列元素个数大于0直接调用私有的dequeue()从对头取出一个元素。 /*** 从当前的取位置提取元素前进位置并发出信号。* 只有在持有锁的情况下才能调用此方法。*/
private E dequeue() {// 断言当前线程持有锁// assert lock.isHeldByCurrentThread();// 断言锁的持有计数为 1// assert lock.getHoldCount() 1;// 断言当前取位置的元素不为 null// assert items[takeIndex] ! null;final Object[] items this.items;SuppressWarnings(unchecked)E e (E) items[takeIndex]; // 从当前取位置提取元素items[takeIndex] null; // 将当前取位置的元素设置为 nullif (takeIndex items.length) takeIndex 0; // takeIndex 增加 1如果达到数组长度则重置为 0count--; // 队列中的元素数量减少 1if (itrs ! null) {itrs.elementDequeued(); // 通知迭代器元素已被移除}notFull.signal(); // 唤醒在 notFull 条件上等待的线程return e; // 返回提取的元素
} 最后操作完毕后释放锁。 2. remove()方法非阻塞移除元素
/*** 从队列中移除指定元素的一个实例如果存在。具体来说移除一个满足 {code o.equals(e)} 的元素* 如果队列中包含一个或多个这样的元素。* 如果队列包含指定的元素或者等价地说如果此调用导致队列发生变化则返回 {code true}。** p基于循环数组的队列中移除内部元素是一个本质上较慢且具有破坏性的操作因此应仅在特殊情况下进行* 理想情况下仅当已知队列未被其他线程访问时才进行。** param o 要从队列中移除的元素如果存在* return 如果此调用导致队列发生变化则返回 {code true}*/
public boolean remove(Object o) {if (o null) return false; // 如果指定的元素为 null直接返回 falsefinal ReentrantLock lock this.lock;lock.lock(); // 获取锁以确保线程安全try {if (count 0) { // 如果队列不为空final Object[] items this.items;for (int i takeIndex, end putIndex, to (i end) ? end : items.length; ; i 0, to end) {for (; i to; i) {if (o.equals(items[i])) { // 如果找到匹配的元素removeAt(i); // 移除该元素return true; // 返回 true 表示队列已更改}}if (to end) break; // 如果已经遍历完所有元素退出循环}}return false; // 没有找到匹配的元素返回 false} finally {lock.unlock(); // 释放锁}
} 3. take()方法阻塞式移除元素
/*** 从队列的头部移除并返回元素如果队列为空则等待直到有元素可用。** return 从队列头部移除并返回的元素* throws InterruptedException 如果当前线程被中断*/
public E take() throws InterruptedException {final ReentrantLock lock this.lock;lock.lockInterruptibly(); // 使用可中断的锁获取机制try {while (count 0) { // 如果队列为空notEmpty.await(); // 等待 notEmpty 条件信号直到队列中有元素}return dequeue(); // 移除并返回队列头部的元素} finally {lock.unlock(); // 释放锁}
} 4.poll(long timeout, TimeUnit unit)阻塞式移除元素
/*** 从队列的头部移除并返回元素如果队列为空则最多等待指定的时间以等待元素变得可用。** param timeout 最大等待时间* param unit 时间单位* return 如果在指定时间内有元素可用则返回队列头部的元素否则返回 {code null}* throws InterruptedException 如果当前线程被中断*/
public E poll(long timeout, TimeUnit unit) throws InterruptedException {long nanos unit.toNanos(timeout); // 将指定的时间转换为纳秒final ReentrantLock lock this.lock;lock.lockInterruptibly(); // 使用可中断的锁获取机制try {while (count 0) { // 如果队列为空if (nanos 0L) { // 如果剩余等待时间小于等于 0return null; // 返回 null 表示没有元素可用}nanos notEmpty.awaitNanos(nanos); // 等待 notEmpty 条件信号直到队列中有元素或超时}return dequeue(); // 移除并返回队列头部的元素} finally {lock.unlock(); // 释放锁}
} 使用lock. lockInterruptibly()加锁保证只能有一个线程执行出队操作。 如果队列是空的当前线程会被阻塞让出lock锁并在notEmpty条件队列中等待被其他线程唤醒。线程被唤醒后需重新尝试获取锁。 线程在两种情况下会被唤醒。 情况1超时被唤醒并获取到锁线程会重新判断是否有数据可用如果队列仍然是空的直接返回null。情况2被其他线程唤醒并获取到锁线程会重新判断是否有数据可用如果队列仍然是空的并且等待超时直接返回null。 如果队列有数据可用调用私有方法dequeue()从队头移除元素。 最后操作完毕后释放锁。
小结
ArrayBlockingQueue是一个有界阻塞队列在初始化时需要指定容量大小在生产者“生产”数据的速度和消费者“消费”数据速度比较稳定且基本匹配的情况下使用Array-BlockingQueue是不错的选择。否则如果生产者产出数据的速度大于消费者消费的速度且当队列中被填满的情况下会有大量生产线程被阻塞。
ArrayBlockingQueue使用独占锁ReentrantLock来实现线程安全出队和入队操作使用同一个锁对象作用类似于synchronized同步锁同时只能有一个线程进行入队和出队操作。这也就意味着生产者和消费者无法并行操作在并发量一般的场景基本够用在高并发场景下可能会成为性能瓶颈。 LinkedBlockingQueue源码解析
Java Review - 并发编程_LinkedBlockingQueue原理源码剖析
LinkedBlockingQueue是一个基于链表实现的无界阻塞队列。此队列按照先进先出FIFO的原则对元素进行排序。
LinkedBlockingQueue类结构 LinkedBlockingQueue和ArrayBlockingQueue的类图结构是一样的。LinkedBlockingQueue实现了BlockingQueue接口继承自AbstractQueue
LinkedBlockingQueue的主要属性
/*** 队列的容量上限如果没有上限则为 Integer.MAX_VALUE*/
private final int capacity;/*** 当前元素的数量*/
private final AtomicInteger count new AtomicInteger();/*** 链表的头节点。* 不变性条件head.item null*/
transient NodeE head;/*** 链表的尾节点。* 不变性条件last.next null*/
private transient NodeE last;/*** 由 take、poll 等方法持有的锁*/
private final ReentrantLock takeLock new ReentrantLock();/*** 等待 take 操作的等待队列*/
SuppressWarnings(serial) // 实现 Condition 接口的类可能是可序列化的
private final Condition notEmpty takeLock.newCondition();/*** 由 put、offer 等方法持有的锁*/
private final ReentrantLock putLock new ReentrantLock();/*** 等待 put 操作的等待队列*/
SuppressWarnings(serial) // 实现 Condition 接口的类可能是可序列化的
private final Condition notFull putLock.newCondition();
1capacity字段队列容量默认容量是Integer.MAX_VALUE。表示队列中最多存储元素个数。2count字段AtomicInteger类型原子变量用于存储队列中元素个数默认值0。3head和last字段LinkedBlockingQueue基于单向链表实现head为链表的头节点last为链表的尾节点。4takeLock和putLock两个非公平独占锁ReentrantLock实例分别保证出队和入队操作线程安全。takeLock用来控制同时只能有一个线程从队头移除元素putLock用来控制同时只能有一个线程从队头插入元素。5notEmpty字段是takeLock的一个Condition实例。当队列为空时获取元素的线程会被阻塞并在该队列等待被唤醒。当队列有数据可用时其他线程会调用notEmpty. signal()会唤醒等待中线程。6notFull字段是takeLock的一个Condition实例。当队列满时插入元素的线程会在该队列等待被唤醒。当队列有空闲位置时其他线程会调用notFull. signal()会唤醒等待中线程。 LinkedBlockingQueue的三个构造函数
/*** 创建一个容量为 {link Integer#MAX_VALUE} 的 {code LinkedBlockingQueue}。*/
public LinkedBlockingQueue() {this(Integer.MAX_VALUE); // 调用带有最大容量的构造函数
}/*** 创建一个具有给定固定容量的 {code LinkedBlockingQueue}。** param capacity 此队列的容量* throws IllegalArgumentException 如果 {code capacity} 不大于零*/
public LinkedBlockingQueue(int capacity) {if (capacity 0) throw new IllegalArgumentException(); // 检查容量是否有效this.capacity capacity; // 设置队列的容量last head new NodeE(null); // 初始化链表的头节点和尾节点
}/*** 创建一个容量为 {link Integer#MAX_VALUE} 的 {code LinkedBlockingQueue}初始包含给定集合中的元素* 元素按集合迭代器的遍历顺序添加。** param c 初始包含的元素集合* throws NullPointerException 如果指定的集合或其任何元素为 null*/
public LinkedBlockingQueue(Collection? extends E c) {this(Integer.MAX_VALUE); // 调用带有最大容量的构造函数final ReentrantLock putLock this.putLock;putLock.lock(); // 获取 put 锁确保可见性try {int n 0;for (E e : c) {if (e null)throw new NullPointerException(); // 检查元素是否为 nullif (n capacity)throw new IllegalStateException(Queue full); // 检查队列是否已满enqueue(new NodeE(e)); // 将元素添加到队列中n;}count.set(n); // 设置当前元素的数量} finally {putLock.unlock(); // 释放 put 锁}
}
使用默认构造函数创建LinkedBlockingQueue实例时会创建一个容量为Integer. MAX_VALUE的阻塞队列近似无界指定初始容量的构造函数根据已有集合初始化队列的构造函数
队列的容量为Integer. MAXVALUE遍历集合将集合中所有元素依次入队。如果集合中元素不能为null否则抛出NullPointerException异常集合中容量大于Integer.MAXVALUE会抛出IllegalStateExceptionQueu full异常。
在用集合初始化队列时使用了putLock锁这是为了保证数据的可见性保证在构造器结束后数据修改结果对所有线程可见。
构造完成后LinkedBlockingQueue的初始结构如下 插入元素
插入元素的逻辑很简单用同一个putLock独占锁保证同时只有一个线程执行插入操作如果队列中有空闲位置元素插入队尾。LinkedBlockingQueue同样也提供了4种不同的处理方式offerE e、addE e、putE e、offeretimeunit。
4种方法主体流程差别不大 , 以put方法为例来分析Linked-BlockingQueue插入元素的流程
线程使用该方法向插入元素时如果队列是满的线程会被一直阻塞并进入notFull条件队列等待。否则如果队列不是满的元素就会被插入到队列尾部。
/*** 将指定的元素插入到队列的尾部如果必要的话等待空间可用。** throws InterruptedException 如果当前线程被中断* throws NullPointerException 如果指定的元素为 null*/
public void put(E e) throws InterruptedException {if (e null) throw new NullPointerException(); // 检查元素是否为 nullfinal int c;final NodeE node new NodeE(e); // 创建新节点final ReentrantLock putLock this.putLock;final AtomicInteger count this.count;putLock.lockInterruptibly(); // 使用可中断的锁获取机制try {/** 注意即使 count 未受锁保护它仍然用于等待条件。* 这是因为在此点上 count 只能减少所有其他 put 操作都被锁排除* 并且如果 count 从容量变为其他值我们或其他等待的 put 操作将收到信号。* 类似地其他等待条件中对 count 的使用也是如此。*/while (count.get() capacity) { // 如果队列已满notFull.await(); // 等待 notFull 条件信号直到队列中有空闲空间}enqueue(node); // 将新节点添加到队列中c count.getAndIncrement(); // 增加元素计数if (c 1 capacity) // 如果增加后队列仍未满notFull.signal(); // 通知其他等待的 put 操作} finally {putLock.unlock(); // 释放锁}if (c 0) // 如果之前队列为空signalNotEmpty(); // 通知等待的 take 操作
}
1判断插入元素不能为Null否则抛出NullPointerException异常。2执行putLock. lockInterruptibly()加锁保证同时只能有一个线程执行入队操作。3如果队列是满的当前线程会被阻塞让出putLock锁并在notFull条件队列中等待被其他线程唤醒。如果队列未满调用enqueuee插入元素。5调用count. getAndIncrement递增元素个数并将递增前队列中元素个数存储在局部变量c中。如果入队后队列仍然未满就调用notFull. signal唤醒一个notFull条件队列上等待线程该线程被唤醒后执行插入元素操作。6入队结束执行putLock. unlock()解锁。解锁后其他线程可以抢占putLock进行入队操作。7如果入队前队列是空的就可能有线程因为执行移除元素操作而被阻塞在notEmpty上。所以当前线程入队完毕后需要唤醒notEmpty中一个等待的线程通知它队列上现有有元素可以获取 插入1个元素后结构如图 插入2个元素后结构如图 移除元素
移除元素的逻辑很简单用同一个takeLock独占锁保证同时只有一个线程执行移除操作如果队列中不为空则从队头取出一个元素。LinkedBlockingQueue同样也提供了4种不同的处理方式remove()、poll()、take()、polltimeunit。
4种方法主体流程差别不大这里仅以take方法为例来分析Linked-BlockingQueue移除元素的流程。
线程使用该方法移除元素时如果队列不为空从队头移除一个元素并返回。如果队列是空的当队列为空时线程会被阻塞并进入notEmpty条件队列等待
/*** 从队列的头部移除并返回元素如果队列为空则等待直到有元素可用。** return 从队列头部移除并返回的元素* throws InterruptedException 如果当前线程被中断*/
public E take() throws InterruptedException {final E x;final int c;final AtomicInteger count this.count;final ReentrantLock takeLock this.takeLock;takeLock.lockInterruptibly(); // 使用可中断的锁获取机制try {while (count.get() 0) { // 如果队列为空notEmpty.await(); // 等待 notEmpty 条件信号直到队列中有元素}x dequeue(); // 从队列中移除并返回头部元素c count.getAndDecrement(); // 减少元素计数if (c 1) // 如果减少后队列中仍有多个元素notEmpty.signal(); // 通知其他等待的 take 操作} finally {takeLock.unlock(); // 释放锁}if (c capacity) // 如果之前队列已满signalNotFull(); // 通知等待的 put 操作return x; // 返回移除的元素
} 使用takeLock. lockInterruptibly()加锁保证只能有一个线程执行出队操作。 如果队列是空的当前线程会被阻塞让出takeLock锁并在notEmpty条件队列中等待被其他线程唤醒。线程被唤醒后需重新尝试获取锁。 线程在两种情况下会被唤醒。 如果队列有数据可用调用私有方法dequeue()从队头移除元素。调用count. getAndDecrement递减元素个数并将递减前队列中元素个数存储在局部变量c中。如果入队后队列仍然不为空就调用notEmpty. signal唤醒一个notEmpty条件队列上等待线程该线程被唤醒后执行移除元素操作。 元素出队结束执行takeLock. unlock()解锁。解锁后其他线程可以抢占takeLock进行出队操作。 如果出队前队列是满的就可能有线程因为执行插入元素操作而被阻塞在notFull上。所以当前元素出队完毕后需要唤醒notFull中一个等待的线程通知它可以继续插入元素了。 移除1个元素后结构如图 最终状态如图 小结
LinkedBlockingQueue是一个基于链表的阻塞队列如果在初始化时没有指定其容量它会默认一个类似无界的容量。如果当生产者产出数据的速度远大于消费者消费的速度时LinkedBlockingQueue会缓存大量的数据这时系统内存可能被消耗殆尽。
LinkedBlockingQueue和ArrayBlockingQueue不同的是它使用两把独立的锁takeLock和putLock来保证出队和入队操作线程安全。这样入队和出队之间可以真正的做到并发执行同时可以有一个线程进行入队操作另一个线程进行出队操作这样比ArrayBlockingQueue提升了2倍的并发效率 ArrayBlockingQueue vs LinkedBlockingQueue 小结
LinkedBlockingQueue和ArrayBlockingQueue两个队列的主要区别如下 1底层数据结构不同。ArrayBlockingQueue基于数组实现使用数组存储元素。LinkedBlockingQueue基于单链表实现使用链表存储元素。 2队列容量不同。ArrayBlockingQueue构造时必须指定容量且后续不能改变。LinkedBlockingQueue既可以指定大小也可以不指定默认使用一个类似无界的容量Integer. MAX_VALUE。 3ArrayBlockingQueue可以使用公平/非公平锁策略而LinkedBlockingQueue只能使用非公平策略。 ArrayBlockingQueue入队和出队公用一把全局ReentrantLock锁LinkedBlockingQueue出队和入队分别使用独立的锁所以LinkedBlockingQueue的并发性能要比ArrayBlockingQueue好
文章转载自： http://www.morning.hkysq.cn.gov.cn.hkysq.cn http://www.morning.qrzwj.cn.gov.cn.qrzwj.cn http://www.morning.bppml.cn.gov.cn.bppml.cn http://www.morning.zmwd.cn.gov.cn.zmwd.cn http://www.morning.gcjhh.cn.gov.cn.gcjhh.cn http://www.morning.xfxqj.cn.gov.cn.xfxqj.cn http://www.morning.fhqsm.cn.gov.cn.fhqsm.cn http://www.morning.kjyhh.cn.gov.cn.kjyhh.cn http://www.morning.rwzqn.cn.gov.cn.rwzqn.cn http://www.morning.lmknf.cn.gov.cn.lmknf.cn http://www.morning.xstfp.cn.gov.cn.xstfp.cn http://www.morning.gtdf.cn.gov.cn.gtdf.cn http://www.morning.jrksk.cn.gov.cn.jrksk.cn http://www.morning.lbhck.cn.gov.cn.lbhck.cn http://www.morning.nssjy.cn.gov.cn.nssjy.cn http://www.morning.ctqlq.cn.gov.cn.ctqlq.cn http://www.morning.kgkph.cn.gov.cn.kgkph.cn http://www.morning.kgphc.cn.gov.cn.kgphc.cn http://www.morning.ytbr.cn.gov.cn.ytbr.cn http://www.morning.shawls.com.cn.gov.cn.shawls.com.cn http://www.morning.wsxxq.cn.gov.cn.wsxxq.cn http://www.morning.lfqnk.cn.gov.cn.lfqnk.cn http://www.morning.xnzmc.cn.gov.cn.xnzmc.cn http://www.morning.qrcsb.cn.gov.cn.qrcsb.cn http://www.morning.pcgmw.cn.gov.cn.pcgmw.cn http://www.morning.litao4.cn.gov.cn.litao4.cn http://www.morning.fndfn.cn.gov.cn.fndfn.cn http://www.morning.bpwz.cn.gov.cn.bpwz.cn http://www.morning.clhyj.cn.gov.cn.clhyj.cn http://www.morning.rnnwd.cn.gov.cn.rnnwd.cn http://www.morning.bkylg.cn.gov.cn.bkylg.cn http://www.morning.ndxrm.cn.gov.cn.ndxrm.cn http://www.morning.bojkosvit.com.gov.cn.bojkosvit.com http://www.morning.zwpzy.cn.gov.cn.zwpzy.cn http://www.morning.jpfpc.cn.gov.cn.jpfpc.cn http://www.morning.bklkt.cn.gov.cn.bklkt.cn http://www.morning.kfmlf.cn.gov.cn.kfmlf.cn http://www.morning.bpmtl.cn.gov.cn.bpmtl.cn http://www.morning.llgpk.cn.gov.cn.llgpk.cn http://www.morning.xjnw.cn.gov.cn.xjnw.cn http://www.morning.pnfwd.cn.gov.cn.pnfwd.cn http://www.morning.rxfgh.cn.gov.cn.rxfgh.cn http://www.morning.ymwcs.cn.gov.cn.ymwcs.cn http://www.morning.mqmxg.cn.gov.cn.mqmxg.cn http://www.morning.mmxt.cn.gov.cn.mmxt.cn http://www.morning.hqlnp.cn.gov.cn.hqlnp.cn http://www.morning.wbqt.cn.gov.cn.wbqt.cn http://www.morning.gywfp.cn.gov.cn.gywfp.cn http://www.morning.tytly.cn.gov.cn.tytly.cn http://www.morning.qxnns.cn.gov.cn.qxnns.cn http://www.morning.hnrqn.cn.gov.cn.hnrqn.cn http://www.morning.xqxrm.cn.gov.cn.xqxrm.cn http://www.morning.rhsg.cn.gov.cn.rhsg.cn http://www.morning.bhjyh.cn.gov.cn.bhjyh.cn http://www.morning.kpfds.cn.gov.cn.kpfds.cn http://www.morning.dbqg.cn.gov.cn.dbqg.cn http://www.morning.rfycj.cn.gov.cn.rfycj.cn http://www.morning.rhzzf.cn.gov.cn.rhzzf.cn http://www.morning.pbknh.cn.gov.cn.pbknh.cn http://www.morning.qstjr.cn.gov.cn.qstjr.cn http://www.morning.kzhxy.cn.gov.cn.kzhxy.cn http://www.morning.qcsbs.cn.gov.cn.qcsbs.cn http://www.morning.ymyhg.cn.gov.cn.ymyhg.cn http://www.morning.zhghd.cn.gov.cn.zhghd.cn http://www.morning.pjwrl.cn.gov.cn.pjwrl.cn http://www.morning.fkflc.cn.gov.cn.fkflc.cn http://www.morning.dbylp.cn.gov.cn.dbylp.cn http://www.morning.nzkc.cn.gov.cn.nzkc.cn http://www.morning.ysnbq.cn.gov.cn.ysnbq.cn http://www.morning.jwxnr.cn.gov.cn.jwxnr.cn http://www.morning.plnry.cn.gov.cn.plnry.cn http://www.morning.buyid.com.cn.gov.cn.buyid.com.cn http://www.morning.jtkfm.cn.gov.cn.jtkfm.cn http://www.morning.cyfsl.cn.gov.cn.cyfsl.cn http://www.morning.smpb.cn.gov.cn.smpb.cn http://www.morning.cgthq.cn.gov.cn.cgthq.cn http://www.morning.yrnyz.cn.gov.cn.yrnyz.cn http://www.morning.jcfg.cn.gov.cn.jcfg.cn http://www.morning.tbwsl.cn.gov.cn.tbwsl.cn http://www.morning.lthtp.cn.gov.cn.lthtp.cn
