NioEventLoopGroup.png
1.Server端的程序如下。几乎所有的Netty程序都是先定义一个bossGroup和一个workerGroup，前者用来接收Client请求，后者用来处理Client请求，各司其职。接下来我们来具体分析一下new NioEventLoopGroup();
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.WARN))
.childHandler(new MyServerInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
2.具体流程如下：
- 进入构造函数
public NioEventLoopGroup() {
this(0);
}
- 调用另一个构造函数。在这个构造函数里，传入了线程数为0，null强转为Executor
public NioEventLoopGroup(int nThreads) {
this(nThreads, (Executor) null);
}
- 调用构造函数，注意这里的构造函数传入了
SelectorProvider.provider()，SelectorProvider定义了创建selector、ServerSocketChannel、SocketChannel等方法，采用Java的 Service Provider Interface (SPI) 方式实现。
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);
}
- DefaultSelectStrategyFactory.INSTANCE这个返回一个DefaultSelectStrategyFactory对象，构造方法被私有化，是单例模式,调用newSelectStrategy() 方法，会返回一个DefaultSelectStrategy对象
public final class DefaultSelectStrategyFactory implements SelectStrategyFactory {
public static final SelectStrategyFactory INSTANCE = new DefaultSelectStrategyFactory();
private DefaultSelectStrategyFactory() { }
@Override
public SelectStrategy newSelectStrategy() {
return DefaultSelectStrategy.INSTANCE;
}
}
-
进入了MultithreadEventLoopGroup类的构造函数，在这里确定了创建几个线程。
DEFAULT_EVENT_LOOP_THREADS,获取核心数的2倍的线程数，例如我的电脑4核心，所以开启8个线程:
static { DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt( "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2)); if (logger.isDebugEnabled()) { logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS); } }protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) { super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args); } -
调用了父类的构造函数MultithreadEventExecutorGroup
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) { this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args); }这里有一个细节DefaultEventExecutorChooserFactory.INSTANCE；在这个构造函数，实例化了一个DefaultEventExecutorChooserFactory，这个工厂是用来创建Executor选择器的，是用来返回使用哪个选择器，根据奇数偶数来判定实例化哪个选择器。
@Override public EventExecutorChooser newChooser(EventExecutor[] executors) { if (isPowerOfTwo(executors.length)) { return new PowerOfTwoEventExecutorChooser(executors); } else { return new GenericEventExecutorChooser(executors); } } -
再次调用MultithreadEventExecutorGroup的另一个构造函数，这里做了很多事情
/** * Create a new instance. * * @param nThreads the number of threads that will be used by this instance. * @param executor the Executor to use, or {@code null} if the default should be used. * @param chooserFactory the {@link EventExecutorChooserFactory} to use. * @param args arguments which will passed to each {@link #newChild(Executor, Object...)} call */ protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, 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) { // Let the caller handle the interruption. Thread.currentThread().interrupt(); break; } } } } } chooser = chooserFactory.newChooser(children); 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); }首先判断了线程是否小于0，如果小于0，抛出异常
-
判断executor是不是null，显然肯定是null
if (executor == null) { executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()); }-
我们来看看newDefaultThreadFactory()做了什么，通过一系列构造函数赋值，就可以通过newThread(command)获得一个线程，然后就可以开启线程了
protected ThreadFactory newDefaultThreadFactory() { return new DefaultThreadFactory(getClass()); }public DefaultThreadFactory(Class<?> poolType) { this(poolType, false, Thread.NORM_PRIORITY); }public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) { this(toPoolName(poolType), daemon, priority); }public DefaultThreadFactory(String poolName, boolean daemon, int priority) { this(poolName, daemon, priority, System.getSecurityManager() == null ? Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup()); }public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) { 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; this.threadGroup = threadGroup; } -
我们再看看ThreadPerTaskExecutor做了什么？其实很简单，将得到的ThreadFactory传入构造函数，然后调用执行，就可以啦。
public final class ThreadPerTaskExecutor implements Executor { private final ThreadFactory threadFactory; public ThreadPerTaskExecutor(ThreadFactory threadFactory) { if (threadFactory == null) { throw new NullPointerException("threadFactory"); } this.threadFactory = threadFactory; } @Override public void execute(Runnable command) { threadFactory.newThread(command).start(); } }
-
-
看看下一句做了什么
children[i] = newChild(executor, args);这个类并没有对这个函数进行实现，到这个类的子类去看实现，这个子类是顶层我们使用的NioEventLoopGroup，返回了一个事件循环，而不是事件循环组
protected EventLoop newChild(Executor executor, Object... args) throws Exception { return new NioEventLoop(this, executor, (SelectorProvider) args[0], ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]); }在NioEventLoop的构造函数中如下所示，主要进行了一些赋值：
其中：final SelectorTuple selectorTuple = openSelector();就是获得一个nio的selector，用于Channel的注册。
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) { super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler); if (selectorProvider == null) { throw new NullPointerException("selectorProvider"); } if (strategy == null) { throw new NullPointerException("selectStrategy"); } provider = selectorProvider; final SelectorTuple selectorTuple = openSelector(); selector = selectorTuple.selector; unwrappedSelector = selectorTuple.unwrappedSelector; selectStrategy = strategy; } -
将执行器放入选择器中，用来获相应的执行器
chooser = chooserFactory.newChooser(children);
