Repository navigation
netty eventloop pool
wangkaiwu edited this page Feb 9, 2023
·
1 revision
以 nioEventLoop 为例
public class NioEventLoopGroup extends MultithreadEventLoopGroup {
public NioEventLoopGroup(int nThreads) {
this(nThreads, (Executor) null);
}
public NioEventLoopGroup(int nThreads, Executor executor) {
this(nThreads, executor, SelectorProvider.provider());
}
public NioEventLoopGroup(
int nThreads, Executor executor, final SelectorProvider selectorProvider) {
this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
}
public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
final SelectStrategyFactory selectStrategyFactory) {
super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
}
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
}public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
private final EventExecutor[] children;
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
}
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i++) {
// 调用NioEventLoopGroup的newChild方法
children[i] = newChild(executor, args);
}
}
}public class NioEventLoopGroup extends MultithreadEventLoopGroup {
@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
// queueFactory 是 null
EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
return new NioEventLoop(this, executor, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
}
}public final class NioEventLoop extends SingleThreadEventLoop {
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
EventLoopTaskQueueFactory queueFactory) {
// taskQueue: newTaskQueue(queueFactory)
super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
rejectedExecutionHandler);
}
private static Queue<Runnable> newTaskQueue(
EventLoopTaskQueueFactory queueFactory) {
if (queueFactory == null) {
// 默认进这里
return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);
}
return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
}
private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
// 默认就是 Integer.MAX_VALUE，所以调用 PlatformDependent.<Runnable>newMpscQueue()
return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
: PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
}
}public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
// 默认就是 Integer.MAX_VALUE
protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,
SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));
protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,
RejectedExecutionHandler rejectedExecutionHandler) {
super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);
}
}public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
private final Queue<Runnable> taskQueue;
private final Executor executor;
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
boolean addTaskWakesUp, Queue<Runnable> taskQueue,
RejectedExecutionHandler rejectedHandler) {
super(parent);
this.executor = ThreadExecutorMap.apply(executor, this);
this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");
}
}public final class PlatformDependent {
private static final int MPSC_CHUNK_SIZE = 1024;
private static final class Mpsc {
// 支持 unsafe 的话，就是 true
private static final boolean USE_MPSC_CHUNKED_ARRAY_QUEUE;
static <T> Queue<T> newMpscQueue() {
return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscUnboundedArrayQueue<T>(MPSC_CHUNK_SIZE)
: new MpscUnboundedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE);
}
}
}public class MpscUnboundedArrayQueue<E> extends BaseMpscLinkedArrayQueue<E> {
public MpscUnboundedArrayQueue(int chunkSize) {
super(chunkSize);
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
public BaseMpscLinkedArrayQueue(final int initialCapacity) {
// initialCapacity 默认值 1024，p2capacity 是 1024
int p2capacity = Pow2.roundToPowerOfTwo(initialCapacity);
// 2046
long mask = (p2capacity - 1) << 1;
// buffer 的 size 是 1025
E[] buffer = allocateRefArray(p2capacity + 1);
producerBuffer = buffer;
producerMask = mask;
// 至少初始化的时候，和 producerBuffer 是同一个
consumerBuffer = buffer;
consumerMask = mask;
soProducerLimit(mask);
}
}生产者相关成员变量
abstract class BaseMpscLinkedArrayQueueColdProducerFields<E> extends BaseMpscLinkedArrayQueuePad3<E> {
private final static long P_LIMIT_OFFSET = fieldOffset(BaseMpscLinkedArrayQueueColdProducerFields.class, "producerLimit");
// limit: 2046
private volatile long producerLimit;
// 2046
protected long producerMask;
// size 是 1025
protected E[] producerBuffer;
final void soProducerLimit(long newValue) {
UNSAFE.putOrderedLong(this, P_LIMIT_OFFSET, newValue);
}
}消费者相关成员变量
abstract class BaseMpscLinkedArrayQueueConsumerFields<E> extends BaseMpscLinkedArrayQueuePad2<E> {
private final static long C_INDEX_OFFSET = fieldOffset(BaseMpscLinkedArrayQueueConsumerFields.class, "consumerIndex");
private volatile long consumerIndex;
protected long consumerMask;
protected E[] consumerBuffer;
}public final class ThreadPerTaskExecutor implements Executor {
private final ThreadFactory threadFactory;
public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory");
}
}public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
private final Queue<Runnable> taskQueue;
@Override
public void execute(Runnable task) {
execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));
}
private void execute(Runnable task, boolean immediate) {
addTask(task);
}
protected void addTask(Runnable task) {
offerTask(task);
}
final boolean offerTask(Runnable task) {
return taskQueue.offer(task);
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
@Override
public boolean offer(final E e) {
long mask;
E[] buffer;
long pIndex;
while (true) {
// 生产者的上限，默认是 2046
long producerLimit = lvProducerLimit();
// 当前生产索引，默认是 0
pIndex = lvProducerIndex();
if ((pIndex & 1) == 1) {
// 说明当前正在 resize，自选等待
continue;
}
// 默认是 2046
mask = this.producerMask;
buffer = this.producerBuffer;
if (producerLimit <= pIndex) {
// 说明达到了生产上限
// 先省略，后面会专门解释
//......
}
// 可以写入
// cas 将 pIndex + 2
if (casProducerIndex(pIndex, pIndex + 2)) {
break;
}
}
// pIndex 还是 + 2 之前的值， mask 是 2046 = 0x0111_1111_1110
final long offset = modifiedCalcCircularRefElementOffset(pIndex, mask);
// 赋值到对应位置
soRefElement(buffer, offset, e);
return true;
}
}final class LinkedArrayQueueUtil {
// 16 = 8+4+4，表示 8 字节对象头，4字节 class指针(开启了压缩类指针)，4字节数组大小
public static final long REF_ARRAY_BASE;
// 2 表示元素指针大小是4 ，2^2 = 4。3 表示 2^3 = 8，表示元素指针大小是8字节
public static final int REF_ELEMENT_SHIFT;
static long modifiedCalcCircularRefElementOffset(long index, long mask) {
// index & mask: 表示不希望索引值超限，且忽略最低有效位
// ((index & mask) << (REF_ELEMENT_SHIFT - 1))： 将 index 扩大2倍
// 所以 + 的前面是数组的头大小，后面是计算偏移，可以转化为 index * 4。4 表示 数组元素指针大小
return REF_ARRAY_BASE + ((index & mask) << (REF_ELEMENT_SHIFT - 1));
}
}public final class UnsafeRefArrayAccess {
public static <E> void soRefElement(E[] buffer, long offset, E e) {
// 线程安全的方式对 buffer 的 pIndex 的元素赋值为 e
UNSAFE.putOrderedObject(buffer, offset, e);
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
@Override
public E poll() {
final E[] buffer = consumerBuffer;
// 获取消费索引
final long index = lpConsumerIndex();
final long mask = consumerMask;
// 获取 index 应该在数组中的索引
final long offset = modifiedCalcCircularRefElementOffset(index, mask);
// 去指定位置获取数据
Object e = lvRefElement(buffer, offset);
if (e == null) {
// 消费索引与生产索引不一致
if (index != lvProducerIndex()) {
// 如果两者不一致，那么 e 不可能为空
// 而 此时 e 为空，说明在扩容，稍后解释
// ......
} else {
// 如果没有在扩容，而消费索引和生产索引又一致，则确信没有数据
return null;
}
} else {
return null;
}
if (e == JUMP) {
// 和扩容有关，稍后讨论
// return ......
}
// 由于本容器是单消费，多生产模型，所以只会有一个消费者
// 获取到了数据，将对应位置置空
soRefElement(buffer, offset, null);
// 消费索引增长
soConsumerIndex(index + 2);
return (E) e;
}
}public final class UnsafeRefArrayAccess {
public static <E> E lvRefElement(E[] buffer, long offset) {
return (E) UNSAFE.getObjectVolatile(buffer, offset);
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
@Override
public boolean offer(final E e) {
long mask;
E[] buffer;
long pIndex;
while (true) {
// 生产者的上限，默认是 2046
long producerLimit = lvProducerLimit();
// 当前生产索引，默认是 0
pIndex = lvProducerIndex();
if ((pIndex & 1) == 1) {
// 说明当前正在 resize，自选等待
continue;
}
// 默认是 2046
mask = this.producerMask;
buffer = this.producerBuffer;
if (producerLimit <= pIndex) {
// 说明达到了生产上限
int result = offerSlowPath(mask, pIndex, producerLimit);
switch (result) {
case CONTINUE_TO_P_INDEX_CAS:
// 只是跳出 switch 块
break;
case RETRY:
continue;
case QUEUE_FULL:
return false;
case QUEUE_RESIZE:
// 触发扩容
resize(mask, buffer, pIndex, e, null);
return true;
}
}
// 可以写入
// cas 将 pIndex + 2
if (casProducerIndex(pIndex, pIndex + 2)) {
break;
}
}
// pIndex 还是 + 2 之前的值， mask 是 2046 = 0x0111_1111_1110
final long offset = modifiedCalcCircularRefElementOffset(pIndex, mask);
// 赋值到对应位置
soRefElement(buffer, offset, e);
return true;
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
private int offerSlowPath(long mask, long pIndex, long producerLimit) {
// 记录消费的索引
final long cIndex = lvConsumerIndex();
// 此处 Capacity 和掩码是一致的
long bufferCapacity = getCurrentBufferCapacity(mask);
// 可以查看下面引用的部分对此处的解释
// pIndex - bufferCapacity 就是待写入部分的长度
if (cIndex + bufferCapacity > pIndex) {
// 由于待写入部分小于 cIndex，所以可以安全的进行覆写
// bufferCapacity 可以看做已写入的部分，cIndex 视为可写入部分
// 总和就是写入上限
if (!casProducerLimit(producerLimit, cIndex + bufferCapacity)) {
// 修改失败
return RETRY;
} else {
// 虽然超出了上限，但是buffer存在覆写部分，所以无需扩容
return CONTINUE_TO_P_INDEX_CAS;
}
}
// 如果是无界队列，availableInQueue(pIndex, cIndex) 返回 Integer.MAX_VALUE
else if (availableInQueue(pIndex, cIndex) <= 0) {
// 无界队列不会满
return QUEUE_FULL;
}
// 由于 pIndex 超出的部分 >= cIndex，所以 cIndex 以下的部分不满足生产数量要求了
// 所以需要对 buffer 扩容，通过将 pIndex 变为奇数就可以标记当前需要扩容
else if (casProducerIndex(pIndex, pIndex + 1)) {
// 需要扩容
return QUEUE_RESIZE;
} else {
// 重试
return RETRY;
}
}
}关于 cIndex + bufferCapacity > pIndex，由以下几个共识
- 此时 pIndex >= bufferCapacity
- 小于cIndex 的部分已经是无用部分了（没有线程读取），所以可以用来再次写入或者说覆写
- 复用有个前提，待写入的长度不能超过 cIndex，否则会把 cIndex 以上的部分给覆盖 > 图例说明：-表示已消费部分，+表示生产部分，\ 表示消费索引 / 表示生产索引，竖线表示buffer边界
- | ---\ ++ +++++ | /
- 由此可知，只要生产索引 >= buffer 边界，且消费索引 > 0，则无需扩容
abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
private void resize(long oldMask, E[] oldBuffer, long pIndex, E e, Supplier<E> s) {
// 计算扩容后的长度
int newBufferLength = getNextBufferSize(oldBuffer);
}
}public class MpscUnboundedArrayQueue<E> extends BaseMpscLinkedArrayQueue<E> {
@Override
protected int getNextBufferSize(E[] buffer) {
// 原先的长度是多少，就扩容多少
return length(buffer);
}
}abstract class BaseMpscLinkedArrayQueue<E> extends BaseMpscLinkedArrayQueueColdProducerFields<E>
implements MessagePassingQueue<E>, QueueProgressIndicators {
private static final Object JUMP = new Object();
private void resize(long oldMask, E[] oldBuffer, long pIndex, E e, Supplier<E> s) {
// 默认还是 oldBuffer 的长度
int newBufferLength = getNextBufferSize(oldBuffer);
// 构造新数组
final E[] newBuffer = allocateRefArray(newBufferLength);
// 当前需要往新数组生产数据
producerBuffer = newBuffer;
// mask 是一种高效的求模方式，此处的计算结果与旧mask保持不变
final int newMask = (newBufferLength - 2) << 1;
producerMask = newMask;
// 原位置
final long offsetInOld = modifiedCalcCircularRefElementOffset(pIndex, oldMask);
// 新位置，实际与原位置是一个值
final long offsetInNew = modifiedCalcCircularRefElementOffset(pIndex, newMask);
// 往对应位置写入数据
soRefElement(newBuffer, offsetInNew, e == null ? s.get() : e);// element in new array
// nextArrayOffset 其实就是获取 oldBuffer 最后一个位置的索引
// 写入的数据是 newBuffer 的指针
soRefElement(oldBuffer, nextArrayOffset(oldMask), newBuffer);// buffer linked
final long cIndex = lvConsumerIndex();
// int 最大值
final long availableInQueue = availableInQueue(pIndex, cIndex);
// pIndex + newMask
soProducerLimit(pIndex + Math.min(newMask, availableInQueue));
// 索引直接 + 2
soProducerIndex(pIndex + 2);
// 旧buffer的位置放置跳转标记
soRefElement(oldBuffer, offsetInOld, JUMP);
}
private static long nextArrayOffset(long mask) {
return modifiedCalcCircularRefElementOffset(mask + 2, Long.MAX_VALUE);
}
}public final class UnsafeRefArrayAccess {
public static <E> E[] allocateRefArray(int capacity) {
return (E[]) new Object[capacity];
}
}是一个多生产单消费的队列
- 当生产超出边界时，若超出部分小于 cIndex，则可以继续复用 cIndex 以下部分用于生产
- 如果不满足复用条件，会触发扩容。做法是在当前 pIndex 对应的 oldBuffer 位置设置 JUMP，并且把 oldBuffer 的末尾设置为 newBuffer 的指针。待生产数据直接写入 newBuffer
- 如果消费到了一个 JUMP 对象，则读取 buffer 末尾的指针，获取 newBuffer 对象。此时 oldBuffer 直接废弃，不会再使用。而且会把 newBuffer 的指针设置为 BUFFER_CONSUMED，表示本队列已废弃，防止重复扩容