类结构
NioEventLoopGroup
继承自MultithreadEventLoopGroup
,多提供了两个方法setIoRatio
和rebuildSelectors
,一个用于设置NioEventLoop
用于IO处理的时间占比,另一个是重新构建Selectors,来处理epoll空轮询导致CPU100%的bug。这个两个的用处在介绍NioEventLoop
的时候在详细介绍。其它的方法都在接口中有定义,先看下EventExecutorGroup
。
EventExecutorGroup
EventExecutorGroup
继承自ScheduledExecutorService
和Iterable
。这意味着EventExecutorGroup
拥有定时处理任务的能力,同时本身可以迭代。它提供的方法有:
/**
* 是否所有事件执行器都处在关闭途中或关闭完成
*/
boolean isShuttingDown();
/**
* 优雅关闭
*/
Future<?> shutdownGracefully();
Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);
/**
* 返回线程池终止时的异步结果
*/
Future<?> terminationFuture();
void shutdown();
List<Runnable> shutdownNow();
/**
* 返回一个事件执行器
*/
EventExecutor next();
其中shutdown
和shutdownNow
被标记为过时,不建议使用。EventExecutorGroup
还重写ScheduledExecutorService
接口的方法,用于返回自定义的Future
。
EventLoopGroup
EventLoopGroup
继承自EventExecutorGroup
,它和EventExecutorGroup
想比多了注册Channel
和ChannelPromise
,同时重新next方法返回EventLoop
。
MultithreadEventExecutorGroup
创建NioEventLoopGroup
时,最终都会调用MultithreadEventExecutorGroup
的构造方法。
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
// 线程数必须大于0
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}
// 没指定Executor就创建新的Executor
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
// 创建EventExecutor数组
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i++) {
// 创建结果标识
boolean success = false;
try {
// 创建EventExecutor对象
children[i] = newChild(executor, args);
// 设置创建成功
success = true;
} catch (Exception e) {
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
// 创建失败,关闭所有已创建的EventExecutor
if (!success) {
// 关闭所有已创建的EventExecutor
for (int j = 0; j < i; j++) {
children[j].shutdownGracefully();
}
// 确保所有已创建的EventExecutor已关闭
for (int j = 0; j < i; j++) {
EventExecutor e = children[j];
try {
while (!e.isTerminated()) {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
} catch (InterruptedException interrupted) {
// Let the caller handle the interruption.
Thread.currentThread().interrupt();
break;
}
}
}
}
}
// 创建EventExecutor选择器
chooser = chooserFactory.newChooser(children);
// 创建监听器,用于EventExecutor终止时的监听
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
// 当EventExecutor全部关闭时
if (terminatedChildren.incrementAndGet() == children.length) {
// 设置结果,并通知监听器们。
terminationFuture.setSuccess(null);
}
}
};
// 给每个EventExecutor添加上监听器
for (EventExecutor e : children) {
e.terminationFuture().addListener(terminationListener);
}
// 创建只读的EventExecutor集合
Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
}
整个构造方法做的就是EventExecutor
的创建,包括创建的异常处理,成功通知等。AbstractEventExecutorGroup
、MultithreadEventLoopGroup
和NioEventLoopGroup
内部没有特殊之处,就不拓展了。
文中帖的代码注释全在:KAMIJYOUDOUMA, 有兴趣的童鞋可以关注一下。
本篇到此结束,如果读完觉得有收获的话,欢迎点赞、关注、加公众号【贰级天災】,查阅更多精彩历史!!!