Java高并发BlockingQueue重要的实现类详解
作者：入门小站 发布时间：2021-06-20 12:08:24
标签：java,高并发,blockingqueue
ArrayBlockingQueue
有界的阻塞队列,内部是一个数组,有边界的意思是:容量是有限的,必须进行初始化,指定它的容量大小,以先进先出的方式存储数据,最新插入的在对尾,最先移除的对象在头部。
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/** 队列元素 */
final Object[] items;
/** 下一次读取操作的位置, poll, peek or remove */
int takeIndex;
/** 下一次写入操作的位置, offer, or add */
int putIndex;
/** 元素数量 */
int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
* 它采用一个 ReentrantLock 和相应的两个 Condition 来实现。
*/
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
/** 指定大小 */
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/**
* 指定容量大小与指定访问策略
* @param fair 指定独占锁是公平锁还是非公平锁。非公平锁的吞吐量比较高，公平锁可以保证每次都是等待最久的线程获取到锁；
*/
public ArrayBlockingQueue(int capacity, boolean fair) {}
/**
* 指定容量大小、指定访问策略与最初包含给定集合中的元素
* @param c 将此集合中的元素在构造方法期间就先添加到队列中
*/
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {}
}
ArrayBlockingQueue 在生产者放入数据和消费者获取数据,都是共用一个锁对象,由此也意味着两者无法真正并行运行。按照实现原理来分析, ArrayBlockingQueue 完全可以采用分离锁,从而实现生产者和消费者操作的完全并行运行。然而事实上并没有如此,因为 ArrayBlockingQueue 的数据写入已经足够轻巧,以至于引入独立的锁机制,除了给代码带来额外的复杂性外,其在性能上完全占不到任何便宜。
通过构造函数得知,参数 fair 控制对象内部是否采用公平锁,默认采用非公平锁。
items、takeIndex、putIndex、count 等属性并没有使用 volatile 修饰,这是因为访问这些变量(通过方法获取)使用都在锁内,并不存在可见性问题,如 size() 。
另外有个独占锁 lock 用来对出入对操作加锁,这导致同时只有一个线程可以访问入队出队。
Put 源码分析
/** 进行入队操作 */
public void put(E e) throws InterruptedException {
//e为null，则抛出NullPointerException异常
checkNotNull(e);
//获取独占锁
final ReentrantLock lock = this.lock;
/**
* lockInterruptibly()
* 获取锁定，除非当前线程为interrupted
* 如果锁没有被另一个线程占用并且立即返回，则将锁定计数设置为1。
* 如果当前线程已经保存此锁，则保持计数将递增1，该方法立即返回。
* 如果锁被另一个线程保持，则当前线程将被禁用以进行线程调度，并且处于休眠状态
*
*/
lock.lockInterruptibly();
try {
//空队列
while (count == items.length)
//进行条件等待处理
notFull.await();
//入队操作
enqueue(e);
} finally {
//释放锁
lock.unlock();
}
}
/** 真正的入队 */
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();
}
这里由于在操作共享变量前加了锁,所以不存在内存不可见问题,加锁后获取的共享变量都是从主内存中获取的,而不是在CPU缓存或者寄存器里面的值,释放锁后修改的共享变量值会刷新到主内存。
另外这个队列使用循环数组实现,所以在计算下一个元素存放下标时候有些特殊。另外 insert 后调用 notEmpty.signal() ;是为了激活调用 notEmpty.await(); 阻塞后放入 notEmpty 条件队列的线程。
Take 源码分析
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
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;
}
Take 操作和 Put 操作很类似
//该类的迭代器，所有的迭代器共享数据，队列改变会影响所有的迭代器
transient Itrs itrs = null; //其存放了目前所创建的所有迭代器。
/**
* 迭代器和它们的队列之间的共享数据，允许队列元素被删除时更新迭代器的修改。
*/
class Itrs {
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
//队列中数量为0的时候，队列就是空的，会将所有迭代器进行清理并移除
queueIsEmpty();
//takeIndex的下标是0，意味着队列从尾中取完了，又回到头部获取
else if (takeIndex == 0)
takeIndexWrapped();
}
/**
* 当队列为空的时候做的事情
* 1. 通知所有迭代器队列已经为空
* 2. 清空所有的弱引用，并且将迭代器置空
*/
void queueIsEmpty() {}
/**
* 将takeIndex包装成0
* 并且通知所有的迭代器，并且删除已经过期的任何对象（个人理解是置空对象）
* 也直接的说就是在Blocking队列进行出队的时候，进行迭代器中的数据同步，保持队列中的元素和迭代器的元素是一致的。
*/
void takeIndexWrapped() {}
}
Itrs迭代器创建的时机
//从这里知道，在ArrayBlockingQueue对象中调用此方法，才会生成这个对象
//那么就可以理解为，只要并未调用此方法，则ArrayBlockingQueue对象中的Itrs对象则为空
public Iterator<E> iterator() {
return new Itr();
}
private class Itr implements Iterator<E> {
Itr() {
//这里就是生产它的地方
//count等于0的时候，创建的这个迭代器是个无用的迭代器，可以直接移除，进入detach模式。
//否则就把当前队列的读取位置给迭代器当做下一个元素，cursor存储下个元素的位置。
if (count == 0) {
// assert itrs == null;
cursor = NONE;
nextIndex = NONE;
prevTakeIndex = DETACHED;
} else {
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
prevTakeIndex = takeIndex;
nextItem = itemAt(nextIndex = takeIndex);
cursor = incCursor(takeIndex);
if (itrs == null) {
itrs = new Itrs(this);
} else {
itrs.register(this); // in this order
itrs.doSomeSweeping(false);
}
prevCycles = itrs.cycles;
// assert takeIndex >= 0;
// assert prevTakeIndex == takeIndex;
// assert nextIndex >= 0;
// assert nextItem != null;
}
}
}
代码演示
package com.rumenz.task;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @className: BlockingQuqueExample
* @description: TODO 类描述
* @author: mac
* @date: 2021/1/20
**/
public class BlockingQueueExample {
private static volatile Boolean flag=false;
public static void main(String[] args) {
BlockingQueue blockingQueue=new ArrayBlockingQueue(1024);
ExecutorService executorService = Executors.newFixedThreadPool(2);
executorService.execute(()->{
try{
blockingQueue.put(1);
Thread.sleep(2000);
blockingQueue.put(3);
flag=true;
}catch (Exception e){
e.printStackTrace();
}
});
executorService.execute(()->{
try {
while (!flag){
Integer i = (Integer) blockingQueue.take();
System.out.println(i);
}
}catch (Exception e){
e.printStackTrace();
}
});
executorService.shutdown();
}
}
LinkedBlockingQueue
基于链表的阻塞队列,通 ArrayBlockingQueue 类似,其内部也维护这一个数据缓冲队列(该队列由一个链表构成),当生产者往队列放入一个数据时,队列会从生产者手上获取数据,并缓存在队列的内部,而生产者立即返回,只有当队列缓冲区到达最大值容量时(LinkedBlockingQueue可以通过构造函数指定该值),才会阻塞队列,直到消费者从队列中消费掉一份数据,生产者会被唤醒,反之对于消费者这端的处理也基于同样的原理。
LinkedBlockingQueue 之所以能够高效的处理并发数据,还因为其对于生产者和消费者端分别采用了独立的锁来控制数据同步,这也意味着在高并发的情况下生产者和消费者可以并行的操作队列中的数据,以调高整个队列的并发能力。
如果构造一个 LinkedBlockingQueue 对象,而没有指定容量大小, LinkedBlockingQueue 会默认一个类似无限大小的容量 Integer.MAX_VALUE ,这样的话,如果生产者的速度一旦大于消费者的速度,也许还没有等到队列满阻塞产生,系统内存就有可能已经被消耗殆尽了。
LinkedBlockingQueue 是一个使用链表完成队列操作的阻塞队列。链表是单向链表,而不是双向链表。
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
//队列的容量，指定大小或为默认值Integer.MAX_VALUE
private final int capacity;
//元素的数量
private final AtomicInteger count = new AtomicInteger();
//队列头节点，始终满足head.item==null
transient Node<E> head;
//队列的尾节点，始终满足last.next==null
private transient Node<E> last;
/** Lock held by take, poll, etc */
//出队的锁：take, poll, peek 等读操作的方法需要获取到这个锁
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
//当队列为空时，保存执行出队的线程：如果读操作的时候队列是空的，那么等待 notEmpty 条件
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
//入队的锁：put, offer 等写操作的方法需要获取到这个锁
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
//当队列满时，保存执行入队的线程：如果写操作的时候队列是满的，那么等待 notFull 条件
private final Condition notFull = putLock.newCondition();
//传说中的 * 队列
public LinkedBlockingQueue() {}
//传说中的有界队列
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
//传说中的 * 队列
public LinkedBlockingQueue(Collection<? extends E> c){}
/**
* 链表节点类
*/
static class Node<E> {
E item;
/**
* One of:
* - 真正的继任者节点
* - 这个节点，意味着继任者是head.next
* - 空，意味着没有后继者（这是最后一个节点）
*/
Node<E> next;
Node(E x) { item = x; }
}
}
通过其构造函数,得知其可以当做 * 队列也可以当做有界队列来使用。
这里用了两把锁分别是 takeLock 和 putLock ,而 Condition 分别是 notEmpty 和 notFull ,它们是这样搭配的。
takeLock
putLock
从上面的构造函数中可以看到,这里会初始化一个空的头结点,那么第一个元素入队的时候,队列中就会有两个元素。读取元素时,也是获取头结点后面的一个元素。count的计数值不包含这个头结点。
Put源码分析
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/**
* 将指定元素插入到此队列的尾部，如有必要，则等待空间变得可用。
*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// 如果你纠结这里为什么是 -1，可以看看 offer 方法。这就是个标识成功、失败的标志而已。
int c = -1;
//包装成node节点
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
//获取锁定
putLock.lockInterruptibly();
try {
/** 如果队列满，等待 notFull 的条件满足。 */
while (count.get() == capacity) {
notFull.await();
}
//入队
enqueue(node);
//原子性自增
c = count.getAndIncrement();
// 如果这个元素入队后，还有至少一个槽可以使用，调用 notFull.signal() 唤醒等待线程。
// 哪些线程会等待在 notFull 这个 Condition 上呢？
if (c + 1 < capacity)
notFull.signal();
} finally {
//解锁
putLock.unlock();
}
// 如果 c == 0，那么代表队列在这个元素入队前是空的（不包括head空节点），
// 那么所有的读线程都在等待 notEmpty 这个条件，等待唤醒，这里做一次唤醒操作
if (c == 0)
signalNotEmpty();
}
/** 链接节点在队列末尾 */
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
// 入队的代码非常简单，就是将 last 属性指向这个新元素，并且让原队尾的 next 指向这个元素
//last.next = node;
//last = node;
// 这里入队没有并发问题，因为只有获取到 putLock 独占锁以后，才可以进行此操作
last = last.next = node;
}
/**
* 等待PUT信号
* 仅在 take/poll 中调用
* 也就是说：元素入队后，如果需要，则会调用这个方法唤醒读线程来读
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();//唤醒
} finally {
putLock.unlock();
}
}
}
Take源码分析
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
//首先，需要获取到 takeLock 才能进行出队操作
takeLock.lockInterruptibly();
try {
// 如果队列为空，等待 notEmpty 这个条件满足再继续执行
while (count.get() == 0) {
notEmpty.await();
}
//// 出队
x = dequeue();
//count 进行原子减 1
c = count.getAndDecrement();
// 如果这次出队后，队列中至少还有一个元素，那么调用 notEmpty.signal() 唤醒其他的读线程
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
/**
* 出队
*/
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;
}
/**
* Signals a waiting put. Called only from take/poll.
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
}
与 ArrayBlockingQueue 对比
ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于GC的影响还是存在一定的区别。
LinkedBlockingQueue 实现一个线程添加文件对象,四个线程读取文件对象
package concurrent;
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class TestBlockingQueue {
static long randomTime() {
return (long) (Math.random() * 1000);
}
public static void main(String[] args) {
// 能容纳100个文件
final BlockingQueue<File> queue = new LinkedBlockingQueue<File>(100);
// 线程池
final ExecutorService exec = Executors.newFixedThreadPool(5);
final File root = new File("F:\\JavaLib");
// 完成标志
final File exitFile = new File("");
// 读个数
final AtomicInteger rc = new AtomicInteger();
// 写个数
final AtomicInteger wc = new AtomicInteger();
// 读线程
Runnable read = new Runnable() {
public void run() {
scanFile(root);
scanFile(exitFile);
}
public void scanFile(File file) {
if (file.isDirectory()) {
File[] files = file.listFiles(new FileFilter() {
public boolean accept(File pathname) {
return pathname.isDirectory()
|| pathname.getPath().endsWith(".java");
}
});
for (File one : files)
scanFile(one);
} else {
try {
int index = rc.incrementAndGet();
System.out.println("Read0: " + index + " "
+ file.getPath());
queue.put(file);
} catch (InterruptedException e) {
}
}
}
};
exec.submit(read);
// 四个写线程
for (int index = 0; index < 4; index++) {
// write thread
final int NO = index;
Runnable write = new Runnable() {
String threadName = "Write" + NO;
public void run() {
while (true) {
try {
Thread.sleep(randomTime());
int index = wc.incrementAndGet();
File file = queue.take();
// 队列已经无对象
if (file == exitFile) {
// 再次添加"标志"，以让其他线程正常退出
queue.put(exitFile);
break;
}
System.out.println(threadName + ": " + index + " "
+ file.getPath());
} catch (InterruptedException e) {
}
}
}
};
exec.submit(write);
}
exec.shutdown();
}
}
总结
来源：https://segmentfault.com/a/1190000039065915
0
投稿猜你喜欢
1.配置代理系统管理---configure Global Security（全局安全设置）---Tcp port for inbound- 添加记录后获取主键ID，这是一个很常见的需求，特别是在一次前端调用中需要插入多个表的场景。除了添加单条记录时获取主键值，有时候可能需要获取批
- mybatis update并非所有字段需要更新mybatis update需求：更新字段作为参数，未更新字段不传入解决办法<upda
- 一、理解 “ 服务器 / 浏览器 ”沟通流程（3步）第1步：浏览器使用<img src=&qu
- 本节作为主要讲解Spring Data的环境搭建JPA Spring Data :致力于减少数据访问层(DAO)的开发量。开发者唯一要做的就
- 概念Java中的集合就是一种容器，可以容纳不同种类的数据，这些容纳是建立在未知的基础上。优点1.可以动态保存任意多个对象，使用比较方便。2.
- 本文实例讲述了Java抛出异常与自定义异常类。分享给大家供大家参考，具体如下：异常处理常用方法：常用的异常处理方法有：一、try()catc
- 原理简介:zookeeper实现分布式锁的原理就是多个节点同时在一个指定的节点下面创建临时会话顺序节点，谁创建的节点序号最小，谁就获得了锁，
- 修改FeginCilent定义的服务名到指定服务通过覆盖类来修改对应的服务名，这里将所有的FeginClient对应的服务名都修改好。pac
- Spring框架七大模块简单介绍Spring中MVC模块代码详解Spring的WEB模块用于整合Web框架，例如Struts1、Struts
- 开始以前，先认识一下WinForm控件数据绑定的两种形式，简单数据绑定和复杂数据绑定。1. 简单的数据绑定例1using (SqlConne
- 前言 在微信刚流行的时候，在摇一摇还能用来那啥的时候，我也曾深更半夜的拿着手机晃一晃。当时想的最多的就是
- Jenkins集成Sonar过程中遇到的报错1、jenkins中无法添加sonarqube的token凭证因为添加的凭证类型错误，所以无法添
- 1，Java中操作方法：import java.io.*; public class FileInputStreamTest &
- 本文实例为大家分享了Java实现登录和注册的具体代码，供大家参考，具体内容如下登录和注册案例的分析：我们在完成一个需求时，需要面向对象，我们
- Java中的try-catch-finally异常处理一、异常处理异常(Exception):是在运行发生的不正常情况。原始异常处理：if(
- 1.基本介绍代码块又称为初始化块，属于类中的成员（类的一部分），类似于方法，讲逻辑语句封装在方法体中，用{}抱起来；但和方法不同，没有方法名
- 1.例题题目描述迷宫由 n 行 m 列的单元格组成，每个单元格要么是空地，要么是障碍物。其中1表示空地，可以走通，2表示障碍物。给定起点坐标
- 定义强引用是使用最普遍的引用。如果一个对象具有强引用，那垃圾回收器宁愿抛出OOM（OutOfMemoryError）也不会回收它。说明不要被
- 1使用背景在实际项目中其中一部分逻辑可能会因为调用了外部服务或者等待锁等情况下出现不可预料的异常，在这个时候我们可能需要对调用这部分逻辑进行
