大家好,今天我准备死磕NioEventLoopGroup的源码，首先讲下概念，NioEventLoopGroup 它是一个线程池，存放NioEventLoop，一个数组，今天打算先看下这行代码的初始化
EventLoopGroup bossGroup = new NioEventLoopGroup();
一、初始化
1、时序图：
2.类的关系
3.源码说明：
步骤1：
EventLoopGroup bossGroup = new NioEventLoopGroup();
//调用链1
public NioEventLoopGroup() {
this(0);
}
//调用链2
public NioEventLoopGroup(int nThreads) {
this(nThreads, (Executor) null);
}
//调用链3
public NioEventLoopGroup(int nThreads, Executor executor) {
this(nThreads, executor, SelectorProvider.provider());
}
//调用链4
public NioEventLoopGroup(
int nThreads, Executor executor, final SelectorProvider selectorProvider) {
super(nThreads, executor, selectorProvider);
}
如上四个方法是NioEventLoopGroup类的四个构造函数，首先会执行第3行的无参构造函数，然后调用第7行的构造函数，再继续调用第11行的构造函数，这里需要获取一个SelectorProvider,见步骤2，再继续调用第15行的构造函数，这里内部实现调用父类MultithreadEventLoopGroup的构造函数。
步骤2：
public static SelectorProvider provider() {
synchronized (lock) {
if (provider != null)
return provider;
return AccessController.doPrivileged(
new PrivilegedAction<SelectorProvider>() {
public SelectorProvider run() {
if (loadProviderFromProperty())
return provider;
if (loadProviderAsService())
return provider;
provider = sun.nio.ch.DefaultSelectorProvider.create();
return provider;
}
});
}
}这里SelectorProvider的静态方法provider获取一个SelectorProvider，全局唯一的，有加锁，依赖底层系统的实现provider不一样，后续另开专题分析；
步骤3：
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}如上是MultithreadEventLoopGroup的构造函数，内部又调用了父类MultithreadEventExecutorGroup的构造函数，继续往下看；
步骤4：
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
if (!success) {
for (int j = 0; j < i; j ++) {
children[j].shutdownGracefully();
}
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) {
Thread.currentThread().interrupt();
break;
}
}
}
}
}
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
}
};
for (EventExecutor e: children) {
e.terminationFuture().addListener(terminationListener);
}
Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
}如上是核心实现的构造函数，代码比较多，我们分段来看下都做了哪些事？
为了方便查看，把代码又粘贴了一遍
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}判断线程数是否小于等于0，否则抛出异常。
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}第6-8行，会调用子类MultithreadEventLoopGroup的newDefaultThreadFactory方法生成ThreadFactory,这里涉及到java继承初始化的知识，不是调用MultithreadEventExecutorGroup的newDefaultThreadFactory方法，需要注意一下；
//调用链1,此方法是MultithreadEventLoopGroup类的
1 protected ThreadFactory newDefaultThreadFactory() {
return new DefaultThreadFactory(getClass(), Thread.MAX_PRIORITY);
} //调用链2
public DefaultThreadFactory(Class<?> poolType, int priority) {
this(poolType, false, priority);
}
//调用链3
public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) {
this(toPoolName(poolType), daemon, priority);
} 14 //调用链4
private static String toPoolName(Class<?> poolType) {
if (poolType == null) {
throw new NullPointerException("poolType");
}
String poolName;
Package pkg = poolType.getPackage();
if (pkg != null) {
poolName = poolType.getName().substring(pkg.getName().length() + 1);
} else {
poolName = poolType.getName();
} switch (poolName.length()) {
case 0:
return "unknown";
case 1:
return poolName.toLowerCase(Locale.US);
default:
if (Character.isUpperCase(poolName.charAt(0)) && Character.isLowerCase(poolName.charAt(1))) {
return Character.toLowerCase(poolName.charAt(0)) + poolName.substring(1);
} else {
return poolName;
}
}
} //调用链5
public DefaultThreadFactory(String poolName, boolean daemon, int priority) {
if (poolName == null) {
throw new NullPointerException("poolName");
}
if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
throw new IllegalArgumentException(
"priority: " + priority + " (expected: Thread.MIN_PRIORITY <= priority <= Thread.MAX_PRIORITY)");
} prefix = poolName + '-' + poolId.incrementAndGet() + '-';
this.daemon = daemon;
this.priority = priority;
}
如上依次从调用链到调用链5，完成DefaultThreadFactory初始化；
再继续初始化ThreadPerTaskExecutor类
public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
if (threadFactory == null) {
throw new NullPointerException("threadFactory");
}
this.threadFactory = threadFactory;
}这里就是new 一个线程工程；
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
if (!success) {
for (int j = 0; j < i; j ++) {
children[j].shutdownGracefully();
} 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) {
Thread.currentThread().interrupt();
break;
}
}
}
}
}
上面初始化了8个(默认，可修改)NioEventLoop, NioEventLoop另外专题分析；
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
}
};创建了一个监听器
for (EventExecutor e: children) {
e.terminationFuture().addListener(terminationListener);
}这里循环调用增加这个监听器；
Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
这里创建一个LinkedHashSet,在把children中的元素全部加进去，然后调用unmodifiableSet,返回一个只读的Chidren，readOnlyChildren;
好了，暂时先分析这么多。