Netty源码—一、server启动（1）

Netty作为一个Java生态中的网络组件有着举足轻重的位置，各种开源中间件都使用Netty进行网络通信，比如Dubbo、RocketMQ。可以说Netty是对Java NIO的封装，比如ByteBuf、channel等的封装让网络编程更简单。

在介绍Netty服务器启动之前需要简单了解两件事：

1. reactor线程模型
2. linux中的IO多路复用

reactor线程模型

关于reactor线程模型请参考这篇文章，通过不同的配置Netty可以实现对应的三种reactor线程模型

• reactor单线程模型
• reactor多线程模型
• reactor主从多线程模型

 // reactor单线程模型，accept、connect、read、write都在一个线程中执行EventLoopGroup group = new NioEventLoopGroup(1);bootStrap.group(group);// reactor多线程，accept在bossGroup中的一个线程执行，IO操作在workerGroup中的线程执行EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();bootStrap.group(bossGroup , workerGroup);// reactor主从多线程，用来accept连接的是在一个线程池中执行，这个时候需要bind多个port，因为Netty一个bind的port会启动一个线程来acceptEventLoopGroup bossGroup = new NioEventLoopGroup(2);EventLoopGroup workerGroup = new NioEventLoopGroup();bootStrap.group(bossGroup , workerGroup);

注意：本文后面的介绍如无特别说明都是基于reactor多线程模型

linux中的IO多路复用

linux中的网络编程模型也是在不断演变的，下面是依次演变的顺序（具体可参考《UNIX网络编程卷1：套接字联网API》第三版的第4、6章）

accept

阻塞等待连接，接收到新的连接后新起线程来处理接收到的连接，然后在新的线程中阻塞等待新的数据到来

select

根据入参的不同有三种情况

1. 永远等下去，直到监听的描述符有任意的IO事件才返回
2. 等待一段固定时间，如果时间到之前有IO事件则提前返回，否则等待超时后返回
3. 不等待，检查描述符后立即返回，称为轮询

select会返回就绪的文件描述符的个数，需要轮询所有socket，判断每个socket的状态来确定是否有事件、是什么事件

poll

相比较于selectpoll是阻塞等待的，只有有读写事件的时候才会返回，返回的是有读写事件的socket个数，并且将对应的socket的事件置位，自己从所有socket中找到具体的socket

epoll

相比较于poll，epoll可以将只有确实有IO事件的描述符返回，大并发下只有少量活跃连接的情况下使用

较poll的优势

1. 不用开发者重新准备文件描述符集合（较poll入参简单）
2. 无需遍历所有监听的描述符，只要遍历哪些被内核IO事件异步唤醒而加入ready队列的描述符集合

Java NIO在linux的实现就是基于epoll的。epoll的编程模型：

1. 创建socket，socket方法
2. 绑定服务器ip，port，bind方法
3. 监听绑定了ip:port的文件描述符，listen方法
4. 创建epoll句柄（文件描述符），配置最大监听的文件描述符个数，epoll_create方法
5. 配置epoll监听的文件描述符的事件：注册、修改、删除某个文件描述符对应的事件
6. 监听所有已配置的描述符，epoll_wait
7. 有新的事件的时候遍历返回的描述符，处理对应的事件
8. 如果是来自客户端的连接，则将accept到的文件描述符注册到epoll中
9. 如果是读写事件则分别处理

注意：Netty封装的Java NIO是跨平台的，后面还是以linux平台为例来介绍

接下来言归正传，来看看Netty的服务器启动过程做了什么事情。Netty作为一个网络框架，和普通网络编程做的事情基本上一样，对应于上面epoll的编程模型，Netty的启动过程为

1. 初始化线程池，初始化selector
2. 初始化NioServerSocketChannel
3. 绑定服务器ip:port
4. 将NioServerSocketChannel注册到selector中
5. 配置NioServerSocketChannel监听的事件
6. 使用selector.select等待新的IO事件
7. 如果是来自客户端的连接则将NioSocketChannel注册到selector上（如果是新的线程则是新的selector）
8. 如果是普通IO事件则在worker线程中处理

线程池初始化

在介绍NioEventLoopGroup之前先看下NioEventLoop

可以看到NioEventLoop继承自SingleThreadEventExecutor，是一个单线程的executor，在线程中死循环监听IO事件。主要方法有

// 初始化selectorio.netty.channel.nio.NioEventLoop#openSelector// 将channel注册到selectorio.netty.channel.nio.NioEventLoop#register// 监听selector上的事件io.netty.channel.nio.NioEventLoop#select

一个NioEventLoop会初始化一个selector，处理selector上注册的channel。

NioEventLoopGroup从名字上就可以看出来是由多个NioEventLoop组成，类关系图如下

NioEventLoopGroup的重要属性为：

// 包含的EventExecutor数组private final EventExecutor[] children;// 选择哪一个EventExecutor执行task的选择器，不同的选择器有不同的策略private final EventExecutorChooserFactory.EventExecutorChooser chooser;

重要方法有：

// 选择下一个执行任务的线程io.netty.util.concurrent.MultithreadEventExecutorGroup#next// 创建EventLoopio.netty.channel.nio.NioEventLoopGroup#newChild// 在线程池中执行注册channel的任务io.netty.channel.MultithreadEventLoopGroup#register(io.netty.channel.Channel)// 创建默认的threadFactoryio.netty.channel.MultithreadEventLoopGroup#newDefaultThreadFactory

线程池初始化的代码为

EventLoopGroup workerGroup = new NioEventLoopGroup();

如果使用无参的构造方法的话，最后会执行下面这个构造方法，这里面做要做了以下几件事

• 如果executor没有初始化，使用默认的executor初始化
• 初始化线程池中每个EventLoop
• 如果其中一个初始化过程中抛出异常，关闭所有的NioEventLoop

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 { ???????????// 创建EventLoop ???????????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，决定选择下一个线程的策略 ???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);}

使用默认参数构造参数的话，上面这个构造方法的入参的值分别是

nThreads

// 默认的线程池大小private static final int DEFAULT_EVENT_LOOP_THREADS;static { ???// 如果配置了io.netty.eventLoopThreads参数的话，先取该参数的值 ???// 如果没有配置上面的参数，则取机器处理器个数的2倍 ???// 如果上面算出的结果小于1则取1 ???DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt( ???????"io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2)); ???if (logger.isDebugEnabled()) { ???????logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS); ???}}// 默认没有指定线程池大小，取DEFAULT_EVENT_LOOP_THREADSprotected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) { ???super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);}

executor

默认没有指定executor，为null

chooserFactory

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) { ???this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);}// io.netty.util.concurrent.DefaultEventExecutorChooserFactory

使用默认的chooser，该类的主要功能是提供选择下一个线程的策略

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory { ???// 单例 ???public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory(); ???private DefaultEventExecutorChooserFactory() { } ???@SuppressWarnings("unchecked") ???@Override ???public EventExecutorChooser newChooser(EventExecutor[] executors) { ???????if (isPowerOfTwo(executors.length)) { ???????????// 如果是2的幂次则使用这个chooser ???????????return new PowerOfTowEventExecutorChooser(executors); ???????} else { ???????????return new GenericEventExecutorChooser(executors); ???????} ???} ???private static boolean isPowerOfTwo(int val) { ???????// 判断一个数是否2的幂，方法很巧妙 ???????return (val & -val) == val; ???} ???private static final class PowerOfTowEventExecutorChooser implements EventExecutorChooser { ???????private final AtomicInteger idx = new AtomicInteger(); ???????private final EventExecutor[] executors; ???????PowerOfTowEventExecutorChooser(EventExecutor[] executors) { ???????????this.executors = executors; ???????} ???????@Override ???????public EventExecutor next() { ???????????// 如果是2的幂次个线程，可以使用位运算计算出下一个选出的线程的index ???????????return executors[idx.getAndIncrement() & executors.length - 1]; ???????} ???} ???private static final class GenericEventExecutorChooser implements EventExecutorChooser { ???????private final AtomicInteger idx = new AtomicInteger(); ???????private final EventExecutor[] executors; ???????GenericEventExecutorChooser(EventExecutor[] executors) { ???????????this.executors = executors; ???????} ???????@Override ???????public EventExecutor next() { ???????????// 使用求余的方法计算出下一个线程的index ???????????return executors[Math.abs(idx.getAndIncrement() % executors.length)]; ???????} ???}}

可以看出上面两个chooser计算出的最终结果是一致的，但是使用位运算更快一点，所以如果是线程池的大小刚好是2的幂次的话使用位运算的chooser。

args

// args[0]，下面方法返回的provider，在linux平台上默认是EPollSelectorProviderjava.nio.channels.spi.SelectorProvider#provider// args[1]，决定eventLoop每次执行select还是执行队列中的任务io.netty.channel.DefaultSelectStrategyFactory// args[2]，等待队列满以后的拒绝策略io.netty.util.concurrent.RejectedExecutionHandlers#REJECT

初始化NioEventLoopGroup过程主要是为了初始化线程池中每一个NioEventLoop，而每一个NioEventLoop包含一个selector。

初始化selector

接着上面说到的初始化NioEventLoop，调用newChild方法来初始化

// io.netty.channel.nio.NioEventLoopGroup#newChildprotected EventLoop newChild(Executor executor, Object... args) throws Exception { ???// 下面这几个参数上面已经介绍过 ???return new NioEventLoop(this, executor, (SelectorProvider) args[0], ???????????????????????????((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);}NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, ????????????SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) { ???// 调用父类构造方法初始化taskQueue，taskQueue的大小取Math.max(16, maxPendingTasks) ???super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler); ???// 校验selectorProvider ???if (selectorProvider == null) { ???????throw new NullPointerException("selectorProvider"); ???} ???// 校验EventLoop每次执行的select策略是否为空 ???if (strategy == null) { ???????throw new NullPointerException("selectStrategy"); ???} ???provider = selectorProvider; ???// 初始化selector ???selector = openSelector(); ???selectStrategy = strategy;}private Selector openSelector() { ???final Selector selector; ???try { ???????// 调用的是sun.nio.ch.EPollSelectorProvider#openSelector ???????// 返回的是sun.nio.ch.EPollSelectorImpl ???????selector = provider.openSelector(); ???} catch (IOException e) { ???????throw new ChannelException("failed to open a new selector", e); ???} ???// 是否使用SelectedSelectionKeySet优化，默认不禁用false ???if (DISABLE_KEYSET_OPTIMIZATION) { ???????return selector; ???} ???// Netty优化过后的 ???final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); ???// 尝试获取SelectorImpl对象，后续会使用反射操作这个类的属性 ???Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() { ???????@Override ???????public Object run() { ???????????try { ???????????????return Class.forName( ???????????????????"sun.nio.ch.SelectorImpl", ???????????????????false, ???????????????????PlatformDependent.getSystemClassLoader()); ???????????} catch (ClassNotFoundException e) { ???????????????return e; ???????????} catch (SecurityException e) { ???????????????return e; ???????????} ???????} ???}); ???// 确保有权限访问该类 ???if (!(maybeSelectorImplClass instanceof Class) || ???????// ensure the current selector implementation is what we can instrument. ???????!((Class<?>) maybeSelectorImplClass).isAssignableFrom(selector.getClass())) { ???????if (maybeSelectorImplClass instanceof Exception) { ???????????Exception e = (Exception) maybeSelectorImplClass; ???????????logger.trace("failed to instrument a special java.util.Set into: {}", selector, e); ???????} ???????return selector; ???} ???final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass; ???Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() { ???????@Override ???????public Object run() { ???????????try { ???????????????// 得到字段selectedKeys ???????????????Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); ???????????????// 得到字段publicSelectedKeys ???????????????Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); ???????????????selectedKeysField.setAccessible(true); ???????????????publicSelectedKeysField.setAccessible(true); ???????????????// 将selectedKeys、publicSelectedKeys均设置为Netty自定义的SelectedSelectionKeySet ???????????????selectedKeysField.set(selector, selectedKeySet); ???????????????publicSelectedKeysField.set(selector, selectedKeySet); ???????????????return null; ???????????} catch (NoSuchFieldException e) { ???????????????return e; ???????????} catch (IllegalAccessException e) { ???????????????return e; ???????????} catch (RuntimeException e) { ???????????????// JDK 9 can throw an inaccessible object exception here; since Netty compiles ???????????????// against JDK 7 and this exception was only added in JDK 9, we have to weakly ???????????????// check the type ???????????????if ("java.lang.reflect.InaccessibleObjectException".equals(e.getClass().getName())) { ???????????????????return e; ???????????????} else { ???????????????????throw e; ???????????????} ???????????} ???????} ???}); ???if (maybeException instanceof Exception) { ???????selectedKeys = null; ???????Exception e = (Exception) maybeException; ???????logger.trace("failed to instrument a special java.util.Set into: {}", selector, e); ???} else { ???????selectedKeys = selectedKeySet; ???????logger.trace("instrumented a special java.util.Set into: {}", selector); ???} ???return selector;}

初始化selector的过程中主要做了几件事：

• 使用平台相关的provider初始化对应的SelectorImpl，这里使用了Java的SPI来加载平台相关的provider，每一种provider又对应一种SelectorImpl
• 如果没有禁用selectedKey优化，Netty会使用自定的SelectedSelectionKeySet替换SelectorImpl的publicSelectedKeys、selectedKeys

对SelectorImpl.selectedKey优化的说明

1. 利用反射将SelectorImpl.selectedKey替换成了SelectedSelectionKeySet，SelectedSelectionKeySet利用数组实现元素存放
2. 在调用select方法的时候如果有事件进来的时候会调用SelectedSelectionKeySet#add，将有IO事件的selectKey添加到keyset中
3. 使用数组遍历（processSelectedKeysOptimized）要比使用set遍历快一些，参考文后第一篇参考文章
4. 在Java9以后这个优化就失效了，因为Java9引入了Jigsaw

接下来看看Selector创建过程，上面调用了EPollSelectorProvider#openSelector来开始初始化selector

public AbstractSelector openSelector() throws IOException { ???// 直接new 一个EPollSelectorImpl ???return new EPollSelectorImpl(this);}// 该构造方法只能是包内使用，供provider来调用EPollSelectorImpl(SelectorProvider sp) throws IOException { ???// 调用父类SelectorImpl的构造方法初始化selectedKeys、publicKeys、publicSelectedKeys ???// 上面已经说过了，如果使用Netty的优化，publicKeys、publicSelectedKey会被替换 ???super(sp); ???// 调用linux的pipe方法，创建一个管道，配置为非阻塞的 ???long pipeFds = IOUtil.makePipe(false); ???// 高32为读文件描述符 ???fd0 = (int) (pipeFds >>> 32); ???// 低32位为写文件描述符 ???fd1 = (int) pipeFds; ???// EPollArrayWrapper包含一系列native方法来调用EPollArrayWrapper.c本地方法 ???pollWrapper = new EPollArrayWrapper(); ???pollWrapper.initInterrupt(fd0, fd1); ???// fdToKey用来保存文件描述符和SelectionKeyImpl的映射 ???fdToKey = new HashMap<>();}EPollArrayWrapper() throws IOException { ???// creates the epoll file descriptor ???// 创建epoll的文件描述符 ???epfd = epollCreate(); ???// the epoll_event array passed to epoll_wait ???int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT; ???pollArray = new AllocatedNativeObject(allocationSize, true); ???pollArrayAddress = pollArray.address(); ???// eventHigh needed when using file descriptors > 64k ???if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE) ???????eventsHigh = new HashMap<>();}

终于看到创建epoll文件描述符相关代码了，上面这个还是看不到究竟调用了哪些本地方法，我们看看相关的c代码

// jdk/src/solaris/native/sun/nio/ch/IOUtil.cJNIEXPORT jlong JNICALLJava_sun_nio_ch_IOUtil_makePipe(JNIEnv *env, jobject this, jboolean blocking){ ???int fd[2]; ???// 打开pipe ???if (pipe(fd) < 0) { ???????JNU_ThrowIOExceptionWithLastError(env, "Pipe failed"); ???????return 0; ???} ???if (blocking == JNI_FALSE) { ???????// 配置管道为非阻塞 ???????if ((configureBlocking(fd[0], JNI_FALSE) < 0) ???????????|| (configureBlocking(fd[1], JNI_FALSE) < 0)) { ???????????JNU_ThrowIOExceptionWithLastError(env, "Configure blocking failed"); ???????????close(fd[0]); ???????????close(fd[1]); ???????????return 0; ???????} ???} ???// 将读写文件描述符放入一个long型中返回 ???return ((jlong) fd[0] << 32) | (jlong) fd[1];}// jdk/src/solaris/native/sun/nio/ch/EPollArrayWrapper.cJNIEXPORT jint JNICALLJava_sun_nio_ch_EPollArrayWrapper_epollCreate(JNIEnv *env, jobject this){ ???/* ????* epoll_create expects a size as a hint to the kernel about how to ????* dimension internal structures. We can‘t predict the size in advance. ????*/ ???// 这里调用linux函数epoll_create创建epoll的文件描述符 ???int epfd = epoll_create(256); ???if (epfd < 0) { ???????JNU_ThrowIOExceptionWithLastError(env, "epoll_create failed"); ???} ???return epfd;}

总结

经过上面说明，现在对于Netty启动过程中线程池的初始化过程和selector初始化过程已经比较清晰了，对于native方法的分析让我们对比linux中epoll编程，对于原理更加清楚。

接下来就是将需要监听的描述符注册到epoll上，对应到Netty就是讲channel注册到selector上，下一篇文章继续写Netty源码—二、server启动（2）

参考

Netty源码分析——Reactor的processSelectedKeys

关于SelectedSelectionKeySet优化的讨论

https://github.com/netty/netty/issues/2363

https://github.com/netty/netty/commit/cd579f75d2b5f236f35bc47f454cc07e50ae8037

Netty源码—一、server启动（1）

原文地址：https://www.cnblogs.com/sunshine-2015/p/9349953.html