netty核心流程（一）：服务端怎样创建连接
为了吸收连接请求， Netty 服务端应该做些什么事情？根据Java NIO 的知识，服务端在预备吸收客户端连接之前做了下面几个工作，我们可以带着问题往下看。
[*]服务端对连接请求是怎样初始化的？
[*]怎样把用户界说的处置处罚逻辑 childHandler 加入到 Netty 的处置处罚流程里？
[*]怎样在 Socket 上绑定一个端口？
[*]怎样把请求连接的 OP_ACCEPT 事件（客户端请求连接的事件）注册到 Selector 上的？
[*]Netty 线程是怎样轮询 ServerSocketChannel 的网络连接事件的？
[*]对于客户端发来的连接请求，服务端是如那边理的？
[*]连接成功创建后，怎样读客户端发来的数据？
我们会根据 Netty 底层的源码来解析上面这几个部分是怎样实现的，源码讲解时我们只讲解主线代码，只要可以大概理解 Netty 大体是怎样实现上述功能就可以了，否则会影响理解。
服务端对连接请求是怎样初始化的？
我们根据上节课 NettyServer 的代码讲起，首先我们先分析一下端口绑定的那行代码：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221550946-2029665855.png
我们对 bind() 方法进行代码追踪，找到方法 doBind()：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221611239-1284716592.png
initAndRegister() 功能是初始化 ServerSocketChannel，然后把对应的网络事件注册到 Selector 的方法，我们具体看看：
// 创建和初始化一个 ServerSocketChannel，并注册到 Selector 轮询复用组件上去。
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 通过工厂类获取 ServerSocketChannel
channel = channelFactory.newChannel();
// 初始化 ServerSocketChannel
init(channel);
} catch (Throwable t) {
if (channel != null) {
channel.unsafe().closeForcibly();
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
// 拿出来之前创建的 EventLoopGroup, 然后把 Channel 注册到 EventLoopGroup上，目的是轮询各种 channel 上的网络事件，
// 我们猜测是不是 让 EvetnLoopGroup 中的独立线程利用一个 Selector 来注册 Channel，并轮询网络事件。
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}可以看到，通过 channelFactory.newChannel() 获得一个 Channel。这个 channelFactory 的实现类很多，这里用的是类 ReflectiveChannelFactory 来实现的，顾名思义是通过反射获得的 Channel。我们再验证下 ReflectiveChannelFactory 是怎样给我们实例化 Channel 的。
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221706634-1283302339.png
可以看到果然是通过反射后得到的 ServerSocketChannel 的实例。
那么，工厂类是怎样知道反射哪个 Channel 呢？这就要看我们上一章写的服务端的代码了。
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221737144-1309560934.png
可以看到，服务端用来吸收请求的类是 NioServerSocketChannel。那么，这个类毕竟有没有封装 ServerSocketChannel 呢？我们看 NioServerSocketChannel 的构造方法就可以了：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221800724-1148458979.png
在这里，我们就真的找到了创建 ServerSocketChannel 实例的代码。也就是说，NioServerSocketChannel 是对 ServerSocketChannel 的封装。到这里，ServerSocketChannel 的实例化就成功了，接着我们看看对 ServerSocketChannel 做了哪些初始化的工作？
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221831029-780979793.png
init(channel) 方法是对 ServerSocketChannel 的初始化。由于是服务端代码，对应的实现类就是 ServerBootstrap，再看它对 init(channel) 的具体实现：
javavoid init(Channel channel) {
// 对ServerSocketChannel 进行相关网络参数的指定。
setChannelOptions(channel, newOptionsArray(), logger);
// 初始化相关的一些属性
setAttributes(channel, newAttributesArray());
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions = newOptionsArray(childOptions);
final Entry<AttributeKey<?>, Object>[] currentChildAttrs = newAttributesArray(childAttrs);
// 第一个拦截器：对网络请求处理链路中加入一个自己内置的一个处理逻辑。初始化了网络请求的处理链路
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
// 第二个拦截器：
pipeline.addLast(handler);
}
// 循环执行下面的任务
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
// 第三个拦截器：加入设定的拦截器
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}可以看到，初始化了一些相关的网络参数，比如上节课 Netty Demo 中 Server 端代码的 SO_BACKLOG 属性。然后初始化一些相关属性。
最后一段紧张的代码是在 Pipeline（拦截器链） 上加入第一个内置的拦截器；第二个拦截器是内部设置的拦截器；然后，通过 eventLoop() 循环处置处罚连接请求，同时还实例化了第三个拦截器 ServerBootstrapAcceptor，然后把第三个拦截器放入拦截器链中，这个过程是启用了一个新的线程来完成的。ServerBootstrapAcceptor 的构造方法的参数包括用户设置的 channel（也就是 NioServerSocketChannel），以及用户设置的 childgroup 线程组，以及处置处罚客户端连接的逻辑 childHandler，最后是一些设置和属性的参数。
初始化就结束了。
怎样把用户界说的处置处罚连接请求的逻辑加入到 Netty 里？
然后，进入了 ServerBootstrapAcceptor：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221914278-492882547.png
这里会看到第四个拦截器。其中，这第四个拦截器才是真正的用户设置的拦截器。随后还会有个监听器，用于拦截器链中所有的拦截器都执行完了以后，再做一些收尾工作。
这段逻辑包涵了 4 个拦截器的创建，比较复杂：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730221940204-76712678.png
现在，服务端连接的初始化已经完成了，但是还没有在 Selector 上注册相应的事件，接下来我们看看这个功能是怎样实现的。
怎样把请求连接的 OP_ACCEPT 事件（客户端请求连接的事件）注册到 Selector 上的？
我们看一下相应的源码：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222010834-1632442598.png
注册网络担当连接请求的事件就在调用 register() 方法里了。
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222032836-135619447.png
由于我们用的是线程组，这里的实现类是 MultithreadEventLoopGroup，其实这里面就是一组 EventLoop。EventLoop 本质上一个线程，每个 EventLoop 会对应一个 Selector。以是，EventLoop 是一个循序处置处罚网络事件的线程。这里通过调用 next() 返回一个线程组里面的一个线程。
接着，我们追踪到 AbstractNioChannel 类中的 doRegister()方法中：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222057232-1757576392.png
在这里，我们就能看到已经把 ServerSocketChannel 以及它对应的 OP_ACCEPT（OP_ACCEPT=0） 事件注册到 Selector 上了。
至此，我们已经把连接事件注册到了对应线程的 Selector 上。
给 ServerSocketChannel 绑定端口
我们回到上一章的 Demo，从 bind() 方法开始看：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222125433-2057479335.png
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222138678-332791584.png
这里就已经把端口绑定到 socket 上了，就不继续深究内部的代码了。
至此，服务端创建连接的预备工作都分析完了。
Netty 线程是怎样轮询 ServerSocketChannel 的网络连接事件的？
首先，我们看看 NioEventLoopGroup 的父类 MultithreadEventLoopGroup 的构造方法：
static {
// 默认的线程数为 CPU 核数的两倍
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);
}MultithreadEventLoopGroup 是一个多线程集合，如果你不设置线程数的话。默认线程数是 CPU 的核数*2。
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222241546-18400094.png
当要增加一个线程时，调用的是 newChild() 方法，我们再看看里面做了什么：
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222305183-176915667.png
我们可以看到，线程组里面的元素是 NioEventLoop 类。
https://img2024.cnblogs.com/blog/2909968/202407/2909968-20240730222320611-464214733.png
很明显，这是一个单线程的线程池。
然后，我们代码跟踪，可以看到往 Selector 里注册网络事件。
private void register0(SelectableChannel ch, int interestOps, NioTask<?> task) {
try {
// 注册网络事件
ch.register(unwrappedSelector, interestOps, task);
} catch (Exception e) {
throw new EventLoopException("failed to register a channel", e);
}
}这里注册的是 OP_ACCEPT 网络事件。
那么，这个线程是在哪里轮询网络事件的呢？这就要找到线程的 run() 方法了
protected void run() {
int selectCnt = 0;
for (;;) {
try {
int strategy;
try {
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
// 通过轮询的方式不断尝试监听新的网络事件
case SelectStrategy.SELECT:
// 设定每次轮询的间隔时间
long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
if (curDeadlineNanos == -1L) {
curDeadlineNanos = NONE; // nothing on the calendar
}
nextWakeupNanos.set(curDeadlineNanos);
try {
if (!hasTasks()) {
// 真正的尝试轮询新的网络事件
strategy = select(curDeadlineNanos);
}
} finally {
nextWakeupNanos.lazySet(AWAKE);
}
// fall through
default:
}通过for(;;)死循环来不断实验发现新的网络事件，如果有就返回。但有个轮询隔断时间来控制执行频率：
private int select(long deadlineNanos) throws IOException { if (deadlineNanos == NONE) { // 轮询网络事件 return selector.select(); } long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L; return timeoutMillis
页:
[1]