824 lines
35 KiB
Markdown
824 lines
35 KiB
Markdown
# 精尽 Netty 源码解析 —— Util 之 MpscUnboundedArrayQueue
|
||
|
||
笔者先把 Netty 主要的内容写完，所以关于 MpscUnboundedArrayQueue 的分享，先放在后续的计划里。
|
||
|
||
> 老艿艿：其实是因为，自己想去研究下 Service Mesh ，所以先简单收个小尾。
|
||
|
||
当然，良心如我，还是为对这块感兴趣的胖友，先准备好了一篇不错的文章：
|
||
|
||
- HMILYYLIMH [《原理剖析（第 012 篇）Netty之无锁队列MpscUnboundedArrayQueue原理分析》](https://www.jianshu.com/p/119a03332619)
|
||
|
||
为避免可能 [《原理剖析（第 012 篇）Netty之无锁队列MpscUnboundedArrayQueue原理分析》](https://www.jianshu.com/p/119a03332619) 被作者删除，笔者这里先复制一份作为备份。
|
||
|
||
# 666. 备份
|
||
|
||
## 一、大致介绍
|
||
|
||
```
|
||
1、了解过netty原理的童鞋，其实应该知道工作线程组的每个子线程都维护了一个任务队列；
|
||
2、细心的童鞋会发现netty的队列是重写了队列的实现方法，覆盖了父类中的LinkedBlockingQueue队列，但是如今却换成了JCTools的一些并发队列，因为JCTools是一款对jdk并发数据结构进行增强的并发工具；
|
||
3、那么问题就来了，现在的netty要用新的队列呢？难道是新的队列确实很高效么？
|
||
4、那么本章节就来和大家分享分析一下Netty新采用的队列之一MpscUnboundedArrayQueue，分析Netty的源码版本为：netty-netty-4.1.22.Final；
|
||
```
|
||
|
||
## 二、回顾预习
|
||
|
||
### 2.1 构造队列
|
||
|
||
```
|
||
1、源码：
|
||
// NioEventLoop.java
|
||
@Override
|
||
protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
|
||
// This event loop never calls takeTask()
|
||
// 由于默认是没有配置io.netty.eventLoop.maxPendingTasks属性值的，所以maxPendingTasks默认值为Integer.MAX_VALUE；
|
||
// 那么最后配备的任务队列的大小也就自然使用无参构造队列方法
|
||
return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
|
||
: PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
|
||
}
|
||
|
||
// PlatformDependent.java
|
||
/**
|
||
* Create a new {@link Queue} which is safe to use for multiple producers (different threads) and a single
|
||
* consumer (one thread!).
|
||
* @return A MPSC queue which may be unbounded.
|
||
*/
|
||
public static <T> Queue<T> newMpscQueue() {
|
||
return Mpsc.newMpscQueue();
|
||
}
|
||
|
||
// Mpsc.java
|
||
static <T> Queue<T> newMpscQueue() {
|
||
return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscUnboundedArrayQueue<T>(MPSC_CHUNK_SIZE)
|
||
: new MpscUnboundedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE);
|
||
}
|
||
|
||
2、通过源码回顾，想必大家已经隐约回忆起之前分析过这段代码，我们在构建工作线程管理组的时候，还需要实例化子线程数组children[]，所以自然就会碰到这段代码；
|
||
|
||
3、这段代码其实就是为了实现一个无锁方式的线程安全队列，总之一句话，效率相当相当的高；
|
||
```
|
||
|
||
### 2.2 何为JCTools？
|
||
|
||
```
|
||
1、JCTools是服务虚拟机并发开发的工具，提供一些JDK没有的并发数据结构辅助开发。
|
||
|
||
2、是一个聚合四种 SPSC/MPSC/SPMC/MPMC 数据变量的并发队列：
|
||
• SPSC：单个生产者对单个消费者（无等待、有界和无界都有实现）
|
||
• MPSC：多个生产者对单个消费者（无锁、有界和无界都有实现）
|
||
• SPMC：单生产者对多个消费者（无锁 有界）
|
||
• MPMC：多生产者对多个消费者（无锁、有界）
|
||
|
||
3、SPSC/MPSC 提供了一个在性能，分配和分配规则之间的平衡的关联数组队列；
|
||
```
|
||
|
||
### 2.3 常用重要的成员属性及方法
|
||
|
||
```
|
||
1、private volatile long producerLimit;
|
||
// 数据链表所分配或者扩展后的容量值
|
||
|
||
2、protected long producerIndex;
|
||
// 生产者指针，每添加一个数据，指针加2
|
||
|
||
3、protected long consumerIndex;
|
||
// 消费者指针，每移除一个数据，指针加2
|
||
|
||
4、private static final int RETRY = 1; // 重新尝试，有可能是因为并发原因，CAS操作指针失败，所以需要重新尝试添加动作
|
||
private static final int QUEUE_FULL = 2; // 队列已满，直接返回false操作
|
||
private static final int QUEUE_RESIZE = 3; // 需要扩容处理，扩容的后的容量值producerLimit一般都是mask的N倍
|
||
// 添加数据时，根据offerSlowPath返回的状态值来做各种处理
|
||
|
||
5、protected E[] producerBuffer;
|
||
// 数据缓冲区，需要添加的数据放在此
|
||
|
||
6、protected long producerMask;
|
||
// 生产者扩充容量值，一般producerMask与consumerMask是一致的，而且需要扩容的数值一般和此值一样
|
||
|
||
7、public boolean offer(final E e)
|
||
// 添加元素
|
||
|
||
8、public E poll()
|
||
// 移除元素
|
||
```
|
||
|
||
### 2.4 数据结构
|
||
|
||
```
|
||
1、如果chunkSize初始化大小为4，则最后显示的数据结构如下：
|
||
E1，E2，。。。，EN：表示具体的元素；
|
||
NBP：表示下一个缓冲区的指针，我采用的是英文的缩写( Next Buffer Pointer)；
|
||
|
||
而且你看着我是拆分开写的，其实每一个NBP指向的就是下面一组缓冲区；
|
||
Buffer1中的NBP其实就是Buffer2的指针引用；
|
||
Buffer2中的NBP其实就是Buffer3的指针引用；
|
||
以此类推。。。
|
||
+------+------+------+------+------+
|
||
| | | | | |
|
||
| E1 | E2 | E3 | JUMP | NBP | Buffer1
|
||
| | | | | |
|
||
+------+------+------+------+------+
|
||
|
||
+------+------+------+------+------+
|
||
| | | | | |
|
||
| E5 | E6 | JUMP | E4 | NBP | Buffer2
|
||
| | | | | |
|
||
+------+------+------+------+------+
|
||
|
||
+------+------+------+------+------+
|
||
| | | | | |
|
||
| E9 | JUMP | E7 | E8 | NBP | Buffer3
|
||
| | | | | |
|
||
+------+------+------+------+------+
|
||
|
||
+------+------+------+------+------+
|
||
| | | | | |
|
||
| JUMP | E10 | E11 | E12 | NBP | Buffer4
|
||
| | | | | |
|
||
+------+------+------+------+------+
|
||
|
||
+------+------+------+------+------+
|
||
| | | | | |
|
||
| E13 | E14 | E15 | JUMP | NBP | Buffer5
|
||
| | | | | |
|
||
+------+------+------+------+------+
|
||
|
||
2、这个数据结构和我们通常所认知的链表是不是有点异样，其实大体还是雷同的，这种数据结构其实也是指针的单项指引罢了；
|
||
```
|
||
|
||
## 三、源码分析MpscUnboundedArrayQueue
|
||
|
||
### 3.1、MpscUnboundedArrayQueue(int)
|
||
|
||
```
|
||
1、源码：
|
||
// MpscUnboundedArrayQueue.java
|
||
public MpscUnboundedArrayQueue(int chunkSize)
|
||
{
|
||
super(chunkSize); // 调用父类的含参构造方法
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueue.java
|
||
/**
|
||
* @param initialCapacity the queue initial capacity. If chunk size is fixed this will be the chunk size.
|
||
* Must be 2 or more.
|
||
*/
|
||
public BaseMpscLinkedArrayQueue(final int initialCapacity)
|
||
{
|
||
// 校验队列容量值，大小必须不小于2
|
||
RangeUtil.checkGreaterThanOrEqual(initialCapacity, 2, "initialCapacity");
|
||
|
||
// 通过传入的参数通过Pow2算法获取大于initialCapacity最近的一个2的n次方的值
|
||
int p2capacity = Pow2.roundToPowerOfTwo(initialCapacity);
|
||
// leave lower bit of mask clear
|
||
long mask = (p2capacity - 1) << 1; // 通过p2capacity计算获得mask值，该值后续将用作扩容的值
|
||
// need extra element to point at next array
|
||
E[] buffer = allocate(p2capacity + 1); // 默认分配一个 p2capacity + 1 大小的数据缓冲区
|
||
producerBuffer = buffer;
|
||
producerMask = mask;
|
||
consumerBuffer = buffer;
|
||
consumerMask = mask;
|
||
// 同时用mask作为初始化队列的Limit值，当生产者指针producerIndex超过该Limit值时就需要做扩容处理
|
||
soProducerLimit(mask); // we know it's all empty to start with
|
||
}
|
||
|
||
// RangeUtil.java
|
||
public static int checkGreaterThanOrEqual(int n, int expected, String name)
|
||
{
|
||
// 要求队列的容量值必须不小于 expected 值，这个 expected 值由上层决定，但是对 MpscUnboundedArrayQueue 而言，expected 为 2；
|
||
// 那么就是说 MpscUnboundedArrayQueue 的值必须不小于 2；
|
||
if (n < expected)
|
||
{
|
||
throw new IllegalArgumentException(name + ": " + n + " (expected: >= " + expected + ')');
|
||
}
|
||
|
||
return n;
|
||
}
|
||
|
||
2、通过调用父类的构造方法，分配了一个数据缓冲区，初始化容量大小，并且容量值不小于2，差不多就这样队列的实例化操作已经完成了；
|
||
```
|
||
|
||
### 3.2、offer(E)
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
@Override
|
||
public boolean offer(final E e)
|
||
{
|
||
if (null == e) // 待添加的元素e不允许为空，否则抛空指针异常
|
||
{
|
||
throw new NullPointerException();
|
||
}
|
||
|
||
long mask;
|
||
E[] buffer;
|
||
long pIndex;
|
||
|
||
while (true)
|
||
{
|
||
long producerLimit = lvProducerLimit(); // 获取当前数据Limit的阈值
|
||
pIndex = lvProducerIndex(); // 获取当前生产者指针位置
|
||
// lower bit is indicative of resize, if we see it we spin until it's cleared
|
||
if ((pIndex & 1) == 1)
|
||
{
|
||
continue;
|
||
}
|
||
// pIndex is even (lower bit is 0) -> actual index is (pIndex >> 1)
|
||
|
||
// mask/buffer may get changed by resizing -> only use for array access after successful CAS.
|
||
mask = this.producerMask;
|
||
buffer = this.producerBuffer;
|
||
// a successful CAS ties the ordering, lv(pIndex) - [mask/buffer] -> cas(pIndex)
|
||
|
||
// assumption behind this optimization is that queue is almost always empty or near empty
|
||
if (producerLimit <= pIndex) // 当阈值小于等于生产者指针位置时，则需要扩容，否则直接通过CAS操作对pIndex做加2处理
|
||
{
|
||
// 通过offerSlowPath返回状态值，来查看怎么来处理这个待添加的元素
|
||
int result = offerSlowPath(mask, pIndex, producerLimit);
|
||
switch (result)
|
||
{
|
||
case CONTINUE_TO_P_INDEX_CAS:
|
||
break;
|
||
case RETRY: // 可能由于并发原因导致CAS失败，那么则再次重新尝试添加元素
|
||
continue;
|
||
case QUEUE_FULL: // 队列已满，直接返回false操作
|
||
return false;
|
||
case QUEUE_RESIZE: // 队列需要扩容操作
|
||
resize(mask, buffer, pIndex, e); // 对队列进行直接扩容操作
|
||
return true;
|
||
}
|
||
}
|
||
|
||
// 能走到这里，则说明当前的生产者指针位置还没有超过阈值，因此直接通过CAS操作做加2处理
|
||
if (casProducerIndex(pIndex, pIndex + 2))
|
||
{
|
||
break;
|
||
}
|
||
}
|
||
// INDEX visible before ELEMENT
|
||
// 获取计算需要添加元素的位置
|
||
final long offset = modifiedCalcElementOffset(pIndex, mask);
|
||
// 在buffer的offset位置添加e元素
|
||
soElement(buffer, offset, e); // release element e
|
||
return true;
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueueProducerFields.java
|
||
@Override
|
||
public final long lvProducerIndex()
|
||
{
|
||
// 通过Unsafe对象调用native方法，获取生产者指针位置
|
||
return UNSAFE.getLongVolatile(this, P_INDEX_OFFSET);
|
||
}
|
||
|
||
// UnsafeRefArrayAccess.java
|
||
/**
|
||
* An ordered store(store + StoreStore barrier) of an element to a given offset
|
||
*
|
||
* @param buffer this.buffer
|
||
* @param offset computed via {@link UnsafeRefArrayAccess#calcElementOffset}
|
||
* @param e an orderly kitty
|
||
*/
|
||
public static <E> void soElement(E[] buffer, long offset, E e)
|
||
{
|
||
// 通过Unsafe对象调用native方法，将元素e设置到buffer缓冲区的offset位置
|
||
UNSAFE.putOrderedObject(buffer, offset, e);
|
||
}
|
||
|
||
2、此方法为添加新的元素对象，当pIndex指针超过阈值producerLimit时则扩容处理，否则直接通过CAS操作添加记录pIndex位置；
|
||
```
|
||
|
||
### 3.3、offerSlowPath(long, long, long)
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
/**
|
||
* We do not inline resize into this method because we do not resize on fill.
|
||
*/
|
||
private int offerSlowPath(long mask, long pIndex, long producerLimit)
|
||
{
|
||
// 获取消费者指针
|
||
final long cIndex = lvConsumerIndex();
|
||
// 获取当前缓冲区的容量值，getCurrentBufferCapacity方法由子类MpscUnboundedArrayQueue实现，默认返回mask值
|
||
long bufferCapacity = getCurrentBufferCapacity(mask);
|
||
|
||
// 如果消费指针加上容量值如果超过了生产指针，那么则会尝试进行扩容处理
|
||
if (cIndex + bufferCapacity > pIndex)
|
||
{
|
||
if (!casProducerLimit(producerLimit, cIndex + bufferCapacity))
|
||
{
|
||
// retry from top
|
||
return RETRY;
|
||
}
|
||
else
|
||
{
|
||
// continue to pIndex CAS
|
||
return CONTINUE_TO_P_INDEX_CAS;
|
||
}
|
||
}
|
||
// full and cannot grow 子类MpscUnboundedArrayQueue默认返回Integer.MAX_VALUE值，所以不会进入此分支
|
||
else if (availableInQueue(pIndex, cIndex) <= 0)
|
||
{
|
||
// offer should return false;
|
||
return QUEUE_FULL;
|
||
}
|
||
// grab index for resize -> set lower bit 尝试扩容队列
|
||
else if (casProducerIndex(pIndex, pIndex + 1))
|
||
{
|
||
// trigger a resize
|
||
return QUEUE_RESIZE;
|
||
}
|
||
else
|
||
{
|
||
// failed resize attempt, retry from top
|
||
return RETRY;
|
||
}
|
||
}
|
||
|
||
// MpscUnboundedArrayQueue.java
|
||
@Override
|
||
protected long getCurrentBufferCapacity(long mask)
|
||
{
|
||
// 获取当前缓冲区的容量值
|
||
return mask;
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueue.java
|
||
final boolean casProducerLimit(long expect, long newValue)
|
||
{
|
||
// 通过CAS尝试对阈值进行修改扩容处理
|
||
return UNSAFE.compareAndSwapLong(this, P_LIMIT_OFFSET, expect, newValue);
|
||
}
|
||
|
||
// MpscUnboundedArrayQueue.java
|
||
@Override
|
||
protected long availableInQueue(long pIndex, long cIndex)
|
||
{
|
||
// 获取可用容量值
|
||
return Integer.MAX_VALUE;
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueueProducerFields.java
|
||
final boolean casProducerIndex(long expect, long newValue)
|
||
{
|
||
// 通过CAS操作更新生产者指针
|
||
return UNSAFE.compareAndSwapLong(this, P_INDEX_OFFSET, expect, newValue);
|
||
}
|
||
|
||
2、该方法主要通过一系列的if...else判断，并结合子类MpscUnboundedArrayQueue的一些重写方法来判断针对该新添加的元素要做何种状态处理；
|
||
```
|
||
|
||
### 3.4、resize(long, E[], long, E)
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
private void resize(long oldMask, E[] oldBuffer, long pIndex, E e)
|
||
{
|
||
// 获取oldBuffer的长度值
|
||
int newBufferLength = getNextBufferSize(oldBuffer);
|
||
// 重新创建新的缓冲区
|
||
final E[] newBuffer = allocate(newBufferLength);
|
||
|
||
producerBuffer = newBuffer; // 将新创建的缓冲区赋值到生产者缓冲区对象上
|
||
final int newMask = (newBufferLength - 2) << 1;
|
||
producerMask = newMask;
|
||
|
||
// 根据oldMask获取偏移位置值
|
||
final long offsetInOld = modifiedCalcElementOffset(pIndex, oldMask);
|
||
// 根据newMask获取偏移位置值
|
||
final long offsetInNew = modifiedCalcElementOffset(pIndex, newMask);
|
||
|
||
// 将元素e设置到新的缓冲区newBuffer的offsetInNew位置处
|
||
soElement(newBuffer, offsetInNew, e);// element in new array
|
||
|
||
// 通过nextArrayOffset(oldMask)计算新的缓冲区将要放置旧的缓冲区的哪个位置
|
||
// 将新的缓冲区newBuffer设置到旧的缓冲区oldBuffer的nextArrayOffset(oldMask)位置处
|
||
// 主要是将oldBuffer中最后一个元素的位置指向新的缓冲区newBuffer
|
||
// 这样就构成了一个单向链表指向的关系
|
||
soElement(oldBuffer, nextArrayOffset(oldMask), newBuffer);// buffer linked
|
||
|
||
// ASSERT code
|
||
final long cIndex = lvConsumerIndex();
|
||
final long availableInQueue = availableInQueue(pIndex, cIndex);
|
||
RangeUtil.checkPositive(availableInQueue, "availableInQueue");
|
||
|
||
// Invalidate racing CASs
|
||
// We never set the limit beyond the bounds of a buffer
|
||
// 重新扩容阈值，因为availableInQueue反正都是Integer.MAX_VALUE值，所以自然就取mask值啦
|
||
// 因此针对MpscUnboundedArrayQueue来说，扩容的值其实就是mask的值的大小
|
||
soProducerLimit(pIndex + Math.min(newMask, availableInQueue));
|
||
|
||
// make resize visible to the other producers
|
||
// 设置生产者指针加2处理
|
||
soProducerIndex(pIndex + 2);
|
||
|
||
// INDEX visible before ELEMENT, consistent with consumer expectation
|
||
|
||
// make resize visible to consumer
|
||
// 用一个空对象来衔接新老缓冲区，凡是在缓冲区中碰到JUMP对象的话，那么就得琢磨着准备着获取下一个缓冲区的数据元素了
|
||
soElement(oldBuffer, offsetInOld, JUMP);
|
||
}
|
||
|
||
// MpscUnboundedArrayQueue.java
|
||
@Override
|
||
protected int getNextBufferSize(E[] buffer)
|
||
{
|
||
// 获取buffer缓冲区的长度
|
||
return length(buffer);
|
||
}
|
||
|
||
// LinkedArrayQueueUtil.java
|
||
static int length(Object[] buf)
|
||
{
|
||
// 直接通过length属性来获取数组的长度
|
||
return buf.length;
|
||
}
|
||
|
||
// CircularArrayOffsetCalculator.java
|
||
@SuppressWarnings("unchecked")
|
||
public static <E> E[] allocate(int capacity)
|
||
{
|
||
// 根据容量值创建数组
|
||
return (E[]) new Object[capacity];
|
||
}
|
||
|
||
2、该方法主要完成新的元素的放置，同时也完成了扩容操作，采用单向链表指针关系，将原缓冲区和新创建的缓冲区衔接起来；
|
||
```
|
||
|
||
### 3.5、poll()
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
/**
|
||
* {@inheritDoc}
|
||
* <p>
|
||
* This implementation is correct for single consumer thread use only.
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
@Override
|
||
public E poll()
|
||
{
|
||
final E[] buffer = consumerBuffer; // 获取缓冲区的数据
|
||
final long index = consumerIndex;
|
||
final long mask = consumerMask;
|
||
|
||
// 根据消费指针与mask来获取当前需要从哪个位置开始来移除元素
|
||
final long offset = modifiedCalcElementOffset(index, mask);
|
||
// 从buffer缓冲区的offset位置获取元素内容
|
||
Object e = lvElement(buffer, offset);// LoadLoad
|
||
if (e == null) // 如果元素为null的话
|
||
{
|
||
// 则再探讨看看消费指针是不是和生产指针是不是相同
|
||
if (index != lvProducerIndex())
|
||
{
|
||
// poll() == null iff queue is empty, null element is not strong enough indicator, so we must
|
||
// check the producer index. If the queue is indeed not empty we spin until element is
|
||
// visible.
|
||
// 若不相同的话，则先尝试从buffer缓冲区的offset位置获取元素先，若获取元素为null则结束while处理
|
||
do
|
||
{
|
||
e = lvElement(buffer, offset);
|
||
}
|
||
while (e == null);
|
||
}
|
||
// 说明消费指针是不是和生产指针是相等的，那么则缓冲区的数据已经被消费完了，直接返回null即可
|
||
else
|
||
{
|
||
return null;
|
||
}
|
||
}
|
||
|
||
// 如果元素为JUMP空对象的话，那么意味着我们就得获取下一缓冲区进行读取数据了
|
||
if (e == JUMP)
|
||
{
|
||
//
|
||
final E[] nextBuffer = getNextBuffer(buffer, mask);
|
||
//
|
||
return newBufferPoll(nextBuffer, index);
|
||
}
|
||
|
||
// 能执行到这里，说明需要移除的元素既不是空的，也不是JUMP空对象，那么则就按照正常处理置空即可
|
||
// 移除元素时，则将buffer缓冲区的offset位置的元素置为空即可
|
||
soElement(buffer, offset, null); // release element null
|
||
// 同时也通过CAS操作增加消费指针的关系，加2操作
|
||
soConsumerIndex(index + 2); // release cIndex
|
||
return (E) e;
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueueProducerFields.java
|
||
@Override
|
||
public final long lvProducerIndex()
|
||
{
|
||
// 通过Unsafe对象调用native方法，获取当前生产者指针值
|
||
return UNSAFE.getLongVolatile(this, P_INDEX_OFFSET);
|
||
}
|
||
|
||
// UnsafeRefArrayAccess.java
|
||
/**
|
||
* A volatile load (load + LoadLoad barrier) of an element from a given offset.
|
||
*
|
||
* @param buffer this.buffer
|
||
* @param offset computed via {@link UnsafeRefArrayAccess#calcElementOffset(long)}
|
||
* @return the element at the offset
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
public static <E> E lvElement(E[] buffer, long offset)
|
||
{
|
||
// 通过Unsafe对象调用native方法，获取buffer缓冲区offset位置的元素
|
||
return (E) UNSAFE.getObjectVolatile(buffer, offset);
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueue.java
|
||
@SuppressWarnings("unchecked")
|
||
private E[] getNextBuffer(final E[] buffer, final long mask)
|
||
{
|
||
// 获取下一个缓冲区的偏移位置值
|
||
final long offset = nextArrayOffset(mask);
|
||
// 从buffer缓冲区的offset位置获取下一个缓冲区数组
|
||
final E[] nextBuffer = (E[]) lvElement(buffer, offset);
|
||
// 获取出来后，同时将buffer缓冲区的offset位置置为空，代表指针已经被取出，原来位置没用了，清空即可
|
||
soElement(buffer, offset, null);
|
||
return nextBuffer;
|
||
}
|
||
|
||
// BaseMpscLinkedArrayQueue.java
|
||
private E newBufferPoll(E[] nextBuffer, long index)
|
||
{
|
||
// 从下一个新的缓冲区中找到需要移除数据的指针位置
|
||
final long offset = newBufferAndOffset(nextBuffer, index);
|
||
// 从newBuffer新的缓冲区中offset位置取出元素
|
||
final E n = lvElement(nextBuffer, offset);// LoadLoad
|
||
if (n == null) // 若取出的元素为空，则直接抛出异常
|
||
{
|
||
throw new IllegalStateException("new buffer must have at least one element");
|
||
}
|
||
// 如果取出的元素不为空，那么先将这个元素原先的位置内容先清空掉
|
||
soElement(nextBuffer, offset, null);// StoreStore
|
||
// 然后通过Unsafe对象调用native方法，修改消费指针的数值偏移加2处理
|
||
soConsumerIndex(index + 2);
|
||
return n;
|
||
}
|
||
|
||
2、该方法主要阐述了该队列是如何的移除数据的；取出的数据如果为JUMP空对象的话，那么则准备从下一个缓冲区获取数据元素，否则还是从当前的缓冲区对象中移除元素，并且更新消费指针；
|
||
```
|
||
|
||
### 3.6、size()
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
@Override
|
||
public final int size()
|
||
{
|
||
// NOTE: because indices are on even numbers we cannot use the size util.
|
||
|
||
/*
|
||
* It is possible for a thread to be interrupted or reschedule between the read of the producer and
|
||
* consumer indices, therefore protection is required to ensure size is within valid range. In the
|
||
* event of concurrent polls/offers to this method the size is OVER estimated as we read consumer
|
||
* index BEFORE the producer index.
|
||
*/
|
||
long after = lvConsumerIndex(); // 获取消费指针
|
||
long size;
|
||
while (true) // 为了防止在获取大小的时候指针发生变化，那么则死循环自旋方式获取大小数值
|
||
{
|
||
final long before = after;
|
||
final long currentProducerIndex = lvProducerIndex(); // 获取生产者指针
|
||
after = lvConsumerIndex(); // 获取消费指针
|
||
|
||
// 如果后获取的消费指针after和之前获取的消费指针before相等的话，那么说明此刻还没有指针变化
|
||
if (before == after)
|
||
{
|
||
// 那么则直接通过生产指针直接减去消费指针，然后向偏移一位，即除以2，得出最后size大小
|
||
size = ((currentProducerIndex - after) >> 1);
|
||
|
||
// 计算完了之后则直接break中断处理
|
||
break;
|
||
}
|
||
|
||
// 若消费指针前后不一致，那么可以说是由于并发原因导致了指针发生了变化；
|
||
// 那么则进行下一次循环继续获取最新的指针值再次进行判断
|
||
}
|
||
// Long overflow is impossible, so size is always positive. Integer overflow is possible for the unbounded
|
||
// indexed queues.
|
||
if (size > Integer.MAX_VALUE)
|
||
{
|
||
return Integer.MAX_VALUE;
|
||
}
|
||
else
|
||
{
|
||
return (int) size;
|
||
}
|
||
}
|
||
|
||
2、获取缓冲区数据大小其实很简单，就是拿着生产指针减去消费指针，但是为了防止并发操作计算错，才用了死循环的方式计算zise值；
|
||
```
|
||
|
||
### 3.7、isEmpty()
|
||
|
||
```
|
||
1、源码：
|
||
// BaseMpscLinkedArrayQueue.java
|
||
@Override
|
||
public final boolean isEmpty()
|
||
{
|
||
// Order matters!
|
||
// Loading consumer before producer allows for producer increments after consumer index is read.
|
||
// This ensures this method is conservative in it's estimate. Note that as this is an MPMC there is
|
||
// nothing we can do to make this an exact method.
|
||
// 这个就简单了，直接判断消费指针和生产指针是不是相等就知道了
|
||
return (this.lvConsumerIndex() == this.lvProducerIndex());
|
||
}
|
||
|
||
2、通过前面我们已经知道了，添加数据的话生产指针在不停的累加操作，而做移除数据的时候消费指针也在不停的累加操作；
|
||
|
||
3、那么这种指针总会有一天会碰面的吧，碰面的那个时候则是数据已经空空如也的时刻；
|
||
```
|
||
|
||
## 四、性能测试
|
||
|
||
```
|
||
1、测试Demo：
|
||
/**
|
||
* 比较队列的消耗情况。
|
||
*
|
||
* @author hmilyylimh
|
||
* ^_^
|
||
* @version 0.0.1
|
||
* ^_^
|
||
* @date 2018/3/30
|
||
*/
|
||
public class CompareQueueCosts {
|
||
|
||
/** 生产者数量 */
|
||
private static int COUNT_OF_PRODUCER = 2;
|
||
|
||
/** 消费者数量 */
|
||
private static final int COUNT_OF_CONSUMER = 1;
|
||
|
||
/** 准备添加的任务数量值 */
|
||
private static final int COUNT_OF_TASK = 1 << 20;
|
||
|
||
/** 线程池对象 */
|
||
private static ExecutorService executorService;
|
||
|
||
public static void main(String[] args) throws Exception {
|
||
|
||
for (int j = 1; j < 7; j++) {
|
||
COUNT_OF_PRODUCER = (int) Math.pow(2, j);
|
||
executorService = Executors.newFixedThreadPool(COUNT_OF_PRODUCER * 2);
|
||
|
||
int basePow = 8;
|
||
int capacity = 0;
|
||
for (int i = 1; i <= 3; i++) {
|
||
capacity = 1 << (basePow + i);
|
||
System.out.print("Producers: " + COUNT_OF_PRODUCER + "\t\t");
|
||
System.out.print("Consumers: " + COUNT_OF_CONSUMER + "\t\t");
|
||
System.out.print("Capacity: " + capacity + "\t\t");
|
||
System.out.print("LinkedBlockingQueue: " + testQueue(new LinkedBlockingQueue<Integer>(capacity), COUNT_OF_TASK) + "s" + "\t\t");
|
||
// System.out.print("ArrayList: " + testQueue(new ArrayList<Integer>(capacity), COUNT_OF_TASK) + "s" + "\t\t");
|
||
// System.out.print("LinkedList: " + testQueue(new LinkedList<Integer>(), COUNT_OF_TASK) + "s" + "\t\t");
|
||
System.out.print("MpscUnboundedArrayQueue: " + testQueue(new MpscUnboundedArrayQueue<Integer>(capacity), COUNT_OF_TASK) + "s" + "\t\t");
|
||
System.out.print("MpscChunkedArrayQueue: " + testQueue(new MpscChunkedArrayQueue<Integer>(capacity), COUNT_OF_TASK) + "s" + "\t\t");
|
||
System.out.println();
|
||
}
|
||
System.out.println();
|
||
|
||
executorService.shutdown();
|
||
}
|
||
}
|
||
|
||
private static Double testQueue(final Collection<Integer> queue, final int taskCount) throws Exception {
|
||
CompletionService<Long> completionService = new ExecutorCompletionService<Long>(executorService);
|
||
|
||
long start = System.currentTimeMillis();
|
||
for (int i = 0; i < COUNT_OF_PRODUCER; i++) {
|
||
executorService.submit(new Producer(queue, taskCount / COUNT_OF_PRODUCER));
|
||
}
|
||
for (int i = 0; i < COUNT_OF_CONSUMER; i++) {
|
||
completionService.submit((new Consumer(queue, taskCount / COUNT_OF_CONSUMER)));
|
||
}
|
||
|
||
for (int i = 0; i < COUNT_OF_CONSUMER; i++) {
|
||
completionService.take().get();
|
||
}
|
||
|
||
long end = System.currentTimeMillis();
|
||
return Double.parseDouble("" + (end - start)) / 1000;
|
||
}
|
||
|
||
private static class Producer implements Runnable {
|
||
private Collection<Integer> queue;
|
||
private int taskCount;
|
||
|
||
public Producer(Collection<Integer> queue, int taskCount) {
|
||
this.queue = queue;
|
||
this.taskCount = taskCount;
|
||
}
|
||
|
||
@Override
|
||
public void run() {
|
||
// Queue队列
|
||
if (this.queue instanceof Queue) {
|
||
Queue<Integer> tempQueue = (Queue<Integer>) this.queue;
|
||
while (this.taskCount > 0) {
|
||
if (tempQueue.offer(this.taskCount)) {
|
||
this.taskCount--;
|
||
} else {
|
||
// System.out.println("Producer offer failed.");
|
||
}
|
||
}
|
||
}
|
||
// List列表
|
||
else if (this.queue instanceof List) {
|
||
List<Integer> tempList = (List<Integer>) this.queue;
|
||
while (this.taskCount > 0) {
|
||
if (tempList.add(this.taskCount)) {
|
||
this.taskCount--;
|
||
} else {
|
||
// System.out.println("Producer offer failed.");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private static class Consumer implements Callable<Long> {
|
||
private Collection<Integer> queue;
|
||
private int taskCount;
|
||
|
||
public Consumer(Collection<Integer> queue, int taskCount) {
|
||
this.queue = queue;
|
||
this.taskCount = taskCount;
|
||
}
|
||
|
||
@Override
|
||
public Long call() {
|
||
// Queue队列
|
||
if (this.queue instanceof Queue) {
|
||
Queue<Integer> tempQueue = (Queue<Integer>) this.queue;
|
||
while (this.taskCount > 0) {
|
||
if ((tempQueue.poll()) != null) {
|
||
this.taskCount--;
|
||
}
|
||
}
|
||
}
|
||
// List列表
|
||
else if (this.queue instanceof List) {
|
||
List<Integer> tempList = (List<Integer>) this.queue;
|
||
while (this.taskCount > 0) {
|
||
if (!tempList.isEmpty() && (tempList.remove(0)) != null) {
|
||
this.taskCount--;
|
||
}
|
||
}
|
||
}
|
||
return 0L;
|
||
}
|
||
}
|
||
}
|
||
|
||
2、指标结果：
|
||
Producers: 2 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 1.399s MpscUnboundedArrayQueue: 0.109s MpscChunkedArrayQueue: 0.09s
|
||
Producers: 2 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 1.462s MpscUnboundedArrayQueue: 0.041s MpscChunkedArrayQueue: 0.048s
|
||
Producers: 2 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 0.281s MpscUnboundedArrayQueue: 0.037s MpscChunkedArrayQueue: 0.082s
|
||
|
||
Producers: 4 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 0.681s MpscUnboundedArrayQueue: 0.085s MpscChunkedArrayQueue: 0.133s
|
||
Producers: 4 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 0.405s MpscUnboundedArrayQueue: 0.094s MpscChunkedArrayQueue: 0.172s
|
||
Producers: 4 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 0.248s MpscUnboundedArrayQueue: 0.107s MpscChunkedArrayQueue: 0.153s
|
||
|
||
Producers: 8 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 1.523s MpscUnboundedArrayQueue: 0.093s MpscChunkedArrayQueue: 0.172s
|
||
Producers: 8 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 0.668s MpscUnboundedArrayQueue: 0.094s MpscChunkedArrayQueue: 0.281s
|
||
Producers: 8 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 0.555s MpscUnboundedArrayQueue: 0.078s MpscChunkedArrayQueue: 0.455s
|
||
|
||
Producers: 16 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 2.676s MpscUnboundedArrayQueue: 0.093s MpscChunkedArrayQueue: 0.753s
|
||
Producers: 16 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 2.135s MpscUnboundedArrayQueue: 0.093s MpscChunkedArrayQueue: 0.792s
|
||
Producers: 16 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 0.944s MpscUnboundedArrayQueue: 0.098s MpscChunkedArrayQueue: 0.64s
|
||
|
||
Producers: 32 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 6.647s MpscUnboundedArrayQueue: 0.078s MpscChunkedArrayQueue: 2.109s
|
||
Producers: 32 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 3.893s MpscUnboundedArrayQueue: 0.095s MpscChunkedArrayQueue: 1.797s
|
||
Producers: 32 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 2.019s MpscUnboundedArrayQueue: 0.109s MpscChunkedArrayQueue: 2.427s
|
||
|
||
Producers: 64 Consumers: 1 Capacity: 512 LinkedBlockingQueue: 26.59s MpscUnboundedArrayQueue: 0.078s MpscChunkedArrayQueue: 3.627s
|
||
Producers: 64 Consumers: 1 Capacity: 1024 LinkedBlockingQueue: 22.566s MpscUnboundedArrayQueue: 0.093s MpscChunkedArrayQueue: 3.047s
|
||
Producers: 64 Consumers: 1 Capacity: 2048 LinkedBlockingQueue: 1.719s MpscUnboundedArrayQueue: 0.093s MpscChunkedArrayQueue: 2.549s
|
||
|
||
3、结果分析(一)：
|
||
通过结果打印耗时可以明显看到MpscUnboundedArrayQueue耗时几乎大多数都是不超过0.1s的，这添加、删除的操作效率不是一般的高，这也难怪人家netty要舍弃自己写的队列框架了；
|
||
|
||
4、结果分析(二)：
|
||
CompareQueueCosts代码里面我将ArrayList、LinkedList注释掉了，那是因为队列数量太大，List的操作太慢，效率低下，所以在大量并发的场景下，大家还是能避免则尽量避免，否则就遭殃了；
|
||
```
|
||
|
||
## 五、总结
|
||
|
||
```
|
||
1、通过底层无锁的Unsafe操作方式实现了多生产者同时访问队列的线程安全模型；
|
||
|
||
2、由于使用锁会造成的线程切换，特别消耗资源，因此使用无锁而是采用CAS的操作方式，虽然会在一定程度上造成CPU使用率过高，但是整体上将效率还是听可观的；
|
||
|
||
3、队列的数据结构是一种单向链表式的结构，通过生产、消费指针来标识添加、移除元素的指针位置，缓冲区与缓冲区之间通过指针指向，避免的数组的复制，较少了大量内存的占用情况；
|
||
``` |