Java并发——阻塞队列集(上)
简介
阻塞队列是一个支持两个附加操做的队列，这两个附加操做支持阻塞的插入和移除方法
①.支持阻塞的插入方法:当队列满时，队列会阻塞插入元素的线程，直至队列不满
②.支持阻塞的移除方法:当队列空时，获取元素的线程会等待队列变为非空java
在阻塞队列不可用时，这两个附加操做提供了4种处理方式，以下node
| 方法/处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
| 插入方法 | add（e） | offer（e） | put（e） | offer（e，time，unit） |
| 移除方法 | remove（） | poll（） | take（） | poll（time，unit） |
| 检查方法 | element（） | peek（） | 不可用 | 不可用 |
阻塞队列
ArrayBlockingQueue：由数组结构组成的有界阻塞队列
LinkedBlockingQueue：由链表结构组成的有界阻塞队列
PriorityBlockingQueue：支持优先级排序的无界阻塞队列
DelayQueue：使用优先级队列实现的无界阻塞队列
SynchronousQueue：不存储元素的阻塞队列
LinkedTransferQueue：由链表结构组成的无界阻塞队列
LinkedBlockingDeque：由链表结构组成的双向阻塞队列
编程
ArrayBlockingQueue
ArrayBlockingQueue是一个用数组实现的有界阻塞队列，队列按照先进先出（FIFO）原则对元素进行排序。默认采用不公平访问，由于公平性一般会下降吞吐量。数组
主要属性
private static final long serialVersionUID = -817911632652898426L;
/** 数组用来维护ArrayBlockingQueue中的元素 */
final Object[] items;
/** 出队首位置索引 */
int takeIndex;
/** 入队末位置索引 */
int putIndex;
/** 元素个数 */
int count;
final ReentrantLock lock;
/** 出队等待队列 */
private final Condition notEmpty;
/** 入队等待队列 */
private final Condition notFull;
复制代码
put
ArrayBlockingQueue提供了不少方法入队：add()、offer()、put()等。咱们以阻塞式方法为主，put()方法其源码以下
缓存
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();
}
}
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
复制代码
当队列满时，会调用Condition的await()方法将线程添加到等待队列中。若队列未满调用enqueue()进行入队操做（全部入队方法最终都将调用该方法在队列尾部插入元素）安全
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
// 入队
items[putIndex] = x;
// 当数组添加满后，从新从0开始
if (++putIndex == items.length)
putIndex = 0;
// 元素个数+1
count++;
// 唤醒出队等待队列中的线程
notEmpty.signal();
}
复制代码
take
出队方法有：poll()、remove()，take()等，take()方法其源码以下并发
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 响应中断式获取同步，若线程被中断会抛出异常
lock.lockInterruptibly();
try {
// 若队列空，将线程添加到notEmpty等待队列中
while (count == 0)
notEmpty.await();
// 获取数据
return dequeue();
} finally {
lock.unlock();
}
}
复制代码
当队列为空，会调用condition的await()方法将线程添加到notEmpty等待队列中，若队列不为空则调用dequeue()获取数据less
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;
// 元素个数-1
count--;
if (itrs != null)
itrs.elementDequeued();
// 通知入队等待队列中的线程
notFull.signal();
return x;
}
复制代码
从源码中能够发现ArrayBlockingQueue经过condition的等待唤醒机制完成可阻塞式的入队和出队函数
LinkedBlockingQueue
LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为 Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序高并发
主要属性
/** 容量 */
private final int capacity;
/** 元素个数 */
private final AtomicInteger count = new AtomicInteger();
/** 头节点 */
transient Node head;
/** 尾节点 */
private transient Node last;
/** 出队锁 */
private final ReentrantLock takeLock = new ReentrantLock();
/** 出队等待队列 */
private final Condition notEmpty = takeLock.newCondition();
/** 入队锁 */
private final ReentrantLock putLock = new ReentrantLock();
/** 入队等待队列 */
private final Condition notFull = putLock.newCondition();
复制代码
从属性上来看LinkedBlockingQueue维护两个锁在入队和出队时保证线程安全，两个锁下降线程因为线程没法获取lock而进入WAITING状态的可能性提升了线程并发执行的效率，而且count属性使用AtomicInteger原子操做类（可能两个线程一个出队一个入队操做count，各自的锁显然起不到用处）
put
public void put(E e) throws InterruptedException {
// 若新增元素为null抛异常
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 node = new Node(e);
final ReentrantLock putLock = this.putLock;
// 获取当前元素个数
final AtomicInteger count = this.count;
// 响应中断式获取锁，若线程被中断会抛出异常
putLock.lockInterruptibly();
try {
// 若当前队列已满，将线程添加到notFull等待队列中
while (count.get() == capacity) {
notFull.await();
}
// 若没有满，进行入队
enqueue(node);
// 元素个数+1
c = count.getAndIncrement();
// 若当前元素个数+1还未到定义的最大容量，则唤醒入队等待队列中的线程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
复制代码
take
public E take() throws InterruptedException {
E x;
int c = -1;
// 获取当前元素个数
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
// 响应中断式获取锁，若线程被中断会抛出异常
takeLock.lockInterruptibly();
try {
// 若当前队列为空，则将线程添加到notEmpty等待队列中
while (count.get() == 0) {
notEmpty.await();
}
// 获取数据
x = dequeue();
// 当前元素个数-1
c = count.getAndDecrement();
// 若队列中还有元素，唤醒阻塞的出队线程
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
复制代码
PriorityBlockingQueue
PriorityBlockingQueue是一个支持优先级的无界阻塞队列，虽然无界但因为资源耗尽，尝试的添加可能会失败（致使OutOfMemoryError ），默认状况下元素采起天然顺序升序排序，也能够经过构造函数来指定Comparator来对元素进行排序，须要注意的是PriorityBlockingQueue不能保证同优先级元素的顺序
主要属性
/** 默认容量 */
private static final int DEFAULT_INITIAL_CAPACITY = 11;
/** 最大容量 */
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
/** 内置数组 */
private transient Object[] queue;
/** 元素个数 */
private transient int size;
/** 比较器，为空则天然排序 */
private transient Comparator comparator;
private final ReentrantLock lock;
/** 出队等待队列 */
private final Condition notEmpty;
/** 用于CAS扩容时用 */
private transient volatile int allocationSpinLock;
private PriorityQueue q;
复制代码
能够发现PriorityBlockingQueue只有一个condition，由于PriorityBlockingQueue是一个无界队列，插入始终成功，也正由于此因此其入队用lock.lock()方法不响应中断，而出队用lock.lockInterruptibly()响应中断式获取锁
put
public void put(E e) {
// 不须要阻塞
offer(e); // never need to block
}
public boolean offer(E e) {
// 判空
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
// 获取锁
lock.lock();
int n, cap;
Object[] array;
// 若大于等于当前数组长度则扩容
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
// 获取比较器
Comparator cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
// 元素个数+1
size = n + 1;
// 唤醒
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
复制代码
tryGrow扩容
private void tryGrow(Object[] array, int oldCap) {
// 必须先释放锁
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
// CAS设置占用
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
// 新容量
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // grow faster if small
(oldCap >> 1));
// 新容量若超过最大值
if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
// 若新容量大于旧容量且当前数组相等，建立新容量数组
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
// CAS设置allocationSpinLock失败，代表有其余线程也正在扩容，让给其余线程处理
if (newArray == null) // back off if another thread is allocating
Thread.yield();
// 获取锁
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
// 数组复制
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
复制代码
从源码中能够发现为了尽量提升并发效率，先释放锁在计算新容量时利用CAS设置allocationSpinLock来保证线程安全，再最后获取锁进行数组复制扩容。扩容完后，根据比较器的排序规则进行新增
siftUpComparable()，比较器comparator为null时采起天然排序调用此方法
private static void siftUpComparable(int k, T x, Object[] array) {
Comparable key = (Comparable) x;
// 若当前元素个数大于0，即队列不为空
while (k > 0) {
// （n - 1） / 2
int parent = (k - 1) >>> 1;
// 获取parent位置上的元素
Object e = array[parent];
// 从队列的最后往上调整堆，直到不小于其父节点为止
if (key.compareTo((T) e) >= 0)
break;
// 若是当前节点小于其父节点，则将其与父节点进行交换，并继续往上访问父节点
array[k] = e;
k = parent;
}
array[k] = key;
}
复制代码
此方法为建堆过程，假定PriorityBlockingQueue内部数组以下:
第二次循环，k=5，parent=2，e=5，5>2交换位置
其主要思路 末位置寻找其父节点，若新增元素小于父节点则将其与父节点进行交换，并继续往上访问父节点，直到大于等于其父节点为止
siftUpUsingComparator()，当比较器不为null，采用指定比较器，调用此方法
private static void siftUpUsingComparator(int k, T x, Object[] array,
Comparator cmp) {
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (cmp.compare(x, (T) e) >= 0)
break;
array[k] = e;
k = parent;
}
array[k] = x;
}
复制代码
take
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
}
return result;
}
复制代码
获取锁后，调用dequeue()
private E dequeue() {
// 若队列为空，返回null
int n = size - 1;
if (n < 0)
return null;
else {
Object[] array = queue;
// 出队元素，首元素
E result = (E) array[0];
// 最后一个元素
E x = (E) array[n];
array[n] = null;
Comparator cmp = comparator;
if (cmp == null)
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
return result;
}
}
复制代码
天然排序处理siftDownComparable()
private static void siftDownComparable(int k, T x, Object[] array,
int n) {
if (n > 0) {
Comparable key = (Comparable)x;
int half = n >>> 1; // loop while a non-leaf
while (k < half) {
// 左节点
int child = (k << 1) + 1; // assume left child is least
Object c = array[child];
// 右节点
int right = child + 1;
if (right < n &&
((Comparable) c).compareTo((T) array[right]) > 0)
c = array[child = right];
if (key.compareTo((T) c) <= 0)
break;
array[k] = c;
k = child;
}
array[k] = key;
}
}
复制代码
指定排序siftDownUsingComparator()
private static void siftDownUsingComparator(int k, T x, Object[] array,
int n,
Comparator cmp) {
if (n > 0) {
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = array[child];
int right = child + 1;
if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
c = array[child = right];
if (cmp.compare(x, (T) c) <= 0)
break;
array[k] = c;
k = child;
}
array[k] = x;
}
}
复制代码
以上面最后一个图为基础出队第一个元素
第二次循环：k=2，child=5，c为图中节点5，right=6，通过子节点比较找出较小值5，5与末位置节点6相比，末位置更大，与左子节点交换位置
第三次循环：k=5，child=11，c为图中节点8，right=12，通过子节点比较找出较小值末位置节点6相比
其主要思路：首位置寻找其子节点，找出两个子节点的较小的与末尾位置节点比较若末尾节点小，则将其置入首位置，不然首位置与较小子节点替换位置，以此略推继续往下找
DelayQueue
DelayQueue是一个支持延时获取元素的无界阻塞队列，队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在建立元素时能够指定多久才能从队列中获取当前元素，只有在延迟期满时才能从队列中提取元素，能够将其应用在缓存、定时任务调度等场景
Delayed接口
DelayQueue队列中的元素必须实现Delayed接口，咱们先看Delayed接口继承关系
从图中咱们能够知道，实现Delayed接口，咱们必须实现其自定义的getDelay()方法以及继承过来的compareTo()方法
主要属性
private final transient ReentrantLock lock = new ReentrantLock();
/** 优先级队列 */
private final PriorityQueue q = new PriorityQueue();
private Thread leader = null;
private final Condition available = lock.newCondition();
复制代码
put
public void put(E e) {
offer(e);
}
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 向PriorityQueue添加元素
q.offer(e);
// 若当前元素
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
复制代码
其添加操做基于PriorityQueue的offer方法
public boolean offer(E e) {
// 判空
if (e == null)
throw new NullPointerException();
// 修改次数
modCount++;
int i = size;
// 判断是否须要扩容
if (i >= queue.length)
grow(i + 1);
// 元素个数+1
size = i + 1;
// 若队列为空，首元素置为e
if (i == 0)
queue[0] = e;
else
siftUp(i, e);
return true;
}
private void siftUp(int k, E x) {
if (comparator != null)
siftUpUsingComparator(k, x);
// 天然排序
else
siftUpComparable(k, x);
}
/**
* 天然排序
*/
private void siftUpComparable(int k, E x) {
Comparable key = (Comparable) x;
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = queue[parent];
if (key.compareTo((E) e) >= 0)
break;
queue[k] = e;
k = parent;
}
queue[k] = key;
}
/**
* 指定比较器
*/
private void siftUpUsingComparator(int k, E x) {
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = queue[parent];
if (comparator.compare(x, (E) e) >= 0)
break;
queue[k] = e;
k = parent;
}
queue[k] = x;
}
复制代码
PriorityQueue的天然排序或指定比较器处理新增操做与PriorityBlockingQueue的逻辑差很少，这里就再也不过多分析，可是从源码咱们发现了modCount，代表PriorityQueue是线程不安全的，可是因为DelayQueue能够依靠ReentrantLock来确保同步安全。新增完后会判断新增元素是否为队列首元素，如果将leader设置为空，并唤醒全部等待线程
take
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 死循环
for (;;) {
// 获取队列首元素，若队列为空返回null
E first = q.peek();
// 若队列为空
if (first == null)
available.await();
else {
// 获取剩余延迟时间
long delay = first.getDelay(NANOSECONDS);
// 若小于0代表已过时，出队
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waiting
// 若leader!= null 代表有其余线程正在操做
if (leader != null)
available.await();
else {
// 不然将leader置为当前线程
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 指定时间等待
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
复制代码
总体出队逻辑再也不多述，来讲下leader和first
感谢
《java并发编程的艺术》
cmsblogs.com/?p=2407
相关文章
- 1. java并发：阻塞队列
- 2. Java 并发--- 阻塞队列
- 3. Java并发——阻塞队列集(下)
- 4. Java并发系列 — 阻塞队列（BlockingQueue）
- 5. Java并发--Java中的阻塞队列
- 6. java并发：非阻塞队列之ConcurrentLinkedQueue
- 7. Java并发中——阻塞队列
- 8. Java并发编程——阻塞队列BlockingQueue
- 9. Java并发编程：阻塞队列BlockingQueue
- 10. (转)Java并发编程：阻塞队列
- 更多相关文章...
- • C# 队列（Queue） - C#教程
- • Rust 并发编程 - RUST 教程
- • ☆技术问答集锦（13）Java Instrument原理
- • 算法总结-归并排序
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
