ARTICLE · 1227050
Netty 服务端启动与事件循环源码剖析
一、ServerBootstrap启动绑定流程
二、NioEventLoop 事件循环机制 三、NioServerSocketChannel Accept 接收流程 四、NioSocketChannel Read 读取流程
本文沿着 Netty 服务端从 bind 到 OP_ACCEPT 注册、再到 NioEventLoop 主循环、Accept 接收与 Read 读取的完整调用链，逐方法拆解源码。先把原生 NIO 的写法与 Netty 封装后的执行路径一一对应，再看 boss 线程与 worker 线程如何切换。
一、ServerBootstrap 启动绑定流程
入口是 io.netty.bootstrap.ServerBootstrap#bind，核心分发在 AbstractBootstrap#doBind。
整体做三件事：先创建并初始化 Channel，再把 Channel 注册到 boss EventLoop，最后由 EventLoop 线程执行端口绑定。
1. doBind 异步注册分支
doBind 先调用 initAndRegister 拿到 regFuture。由于注册是异步的，需要根据是否完成分两条路径：已完成立刻调用 doBind0；未完成则挂一个监听器，等注册线程回调 doBind0。
private ChannelFuture doBind(final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; } if (regFuture.isDone()) { ChannelPromise promise = channel.newPromise(); doBind0(regFuture, channel, localAddress, promise); return promise; } else { final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); regFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { Throwable cause = future.cause(); if (cause != null) { promise.setFailure(cause); } else { promise.registered(); doBind0(regFuture, channel, localAddress, promise); } } }); return promise; }}2. initAndRegister 通道创建注册
initAndRegister 分两步：channelFactory.newChannel 反射创建 NioServerSocketChannel，init 往里塞初始化器；再把 Channel 交给 boss group 注册。
final ChannelFuture initAndRegister() { Channel channel = null; try { channel = channelFactory.newChannel(); init(channel); } catch (Throwable t) { return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE) .setFailure(t); } ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { // 处理异常 } return regFuture;}3. init ChannelPipeline 初始化器装配
ServerBootstrap#init 做三件事：设置 ServerSocketChannel 自己的 options 和 attrs；往 pipeline 尾部塞一个 ChannelInitializer 匿名类；该初始化器在后续注册完成后再把 ServerBootstrapAcceptor 加进 pipeline。
ServerBootstrapAcceptor 是 boss 线程上唯一真正处理连接的 handler，它持有 childGroup、childHandler、childOptions、childAttrs，用于把每个新接入的 NioSocketChannel 装配并注册到 worker 组。
void init(Channel channel) throws Exception { final Map<ChannelOption<?, Object>> options = options0(); synchronized (options) { setChannelOptions(channel, options, logger); } final Map<AttributeKey<?, Object>> attrs = attrs0(); synchronized (attrs) { for (Entry<AttributeKey<?, Object>> e : attrs.entrySet()) { @SuppressWarnings("unchecked") AttributeKey<Object> key = (AttributeKey<Object>) e.getKey(); channel.attr(key).set(e.getValue()); } } ChannelPipeline p = channel.pipeline(); final EventLoopGroup currentChildGroup = childGroup; final ChannelHandler currentChildHandler = childHandler; final Entry<ChannelOption<?, Object>>[] currentChildOptions; final Entry<AttributeKey<?, Object>>[] currentChildAttrs; synchronized (childOptions) { currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0)); } synchronized (childAttrs) { currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0)); } p.addLast(new ChannelInitializer<Channel>() { @Override public void initChannel(final Channel ch) throws Exception { 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)); } }); } });}4. AbstractUnsafe Register 线程切换
group.register 最终走到 AbstractChannel.AbstractUnsafe#register。首次调用时调用方是 main 线程，eventLoop.inEventLoop 为 false，于是通过 eventLoop.execute 提交一个任务，由 boss NIO 线程后续执行 register0。这一行代码完成了 main 到 boss 线程的切换，也顺带启动了 boss 线程。
public final void register(EventLoop eventLoop, final ChannelPromise promise) { // 一些检查 AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop()) { register0(promise); } else { try { eventLoop.execute(new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } }}5. register0 原生通道注册
register0 在 boss 线程上执行。doRegister 把原生 ServerSocketChannel 绑定到 selector，interestOps 此时为 0，附件为 NioServerSocketChannel。
接着 invokeHandlerAddedIfNeeded 触发 ChannelInitializer 回调，safeSetSuccess 让 regFuture 完成，最后 fireChannelRegistered。
此时 ServerSocketChannel 还没 bind，isActive 为 false，因此不会立刻 fireChannelActive，也不会注册 OP_ACCEPT。
private void register0(ChannelPromise promise) { try { if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; doRegister(); neverRegistered = false; registered = true; pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); pipeline.fireChannelRegistered(); if (isActive()) { if (firstRegistration) { pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); }}6. ChannelInitializer initChannel 回调执行
ChannelInitializer 是一次性 handler。initChannel 回调用户传入的初始化逻辑后，在 finally 中把自己从 pipeline 移除，避免重复执行。
private boolean initChannel(ChannelHandlerContext ctx) throws Exception { if (initMap.add(ctx)) { try { initChannel((C) ctx.channel()); } catch (Throwable cause) { exceptionCaught(ctx, cause); } finally { ChannelPipeline pipeline = ctx.pipeline(); if (pipeline.context(this) != null) { pipeline.remove(this); } } return true; } return false;}7. doBind0 绑定任务提交
regFuture 完成后，无论是同步分支还是异步监听器分支，最终都调用 doBind0。它再往 boss EventLoop 提交一个任务，在任务里执行 channel.bind。
private static void doBind0( final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) { channel.eventLoop().execute(new Runnable() { @Override public void run() { if (regFuture.isSuccess()) { channel.bind(localAddress, promise) .addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } else { promise.setFailure(regFuture.cause()); } } });}8. AbstractUnsafe Bind 端口绑定
bind 在 boss 线程上执行。doBind 完成原生 bind 后，wasActive 为 false、isActive 变为 true，于是通过 invokeLater 触发 pipeline.fireChannelActive。
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { assertEventLoop(); if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean wasActive = isActive(); try { doBind(localAddress); } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { invokeLater(new Runnable() { @Override public void run() { pipeline.fireChannelActive(); } }); } safeSetSuccess(promise);}9. NioServerSocketChannel doBind 原生绑定
doBind 直接委托给原生 ServerSocketChannel.bind，传入 backlog。
protected void doBind(SocketAddress localAddress) throws Exception { if (PlatformDependent.javaVersion() >= 7) { javaChannel().bind(localAddress, config.getBacklog()); } else { javaChannel().socket().bind(localAddress, config.getBacklog()); }}10. HeadContext channelActive OP_ACCEPT 注册
fireChannelActive 从 pipeline 尾部往前传播，最终到达 HeadContext。HeadContext.channelActive 调用 readIfIsAutoRead，触发 doBeginRead，把 OP_ACCEPT 加入 interestOps。至此 boss 线程才真正开始关注 accept 事件。
public void channelActive(ChannelHandlerContext ctx) { ctx.fireChannelActive(); readIfIsAutoRead();}protected void doBeginRead() throws Exception { final SelectionKey selectionKey = this.selectionKey; if (!selectionKey.isValid()) { return; } readPending = true; final int interestOps = selectionKey.interestOps(); if ((interestOps & readInterestOp) == 0) { selectionKey.interestOps(interestOps | readInterestOp); }}readInterestOp 在 NioServerSocketChannel 创建时被赋值为 16，即 OP_ACCEPT。
二、NioEventLoop 事件循环机制
NioEventLoop 既要处理 IO 事件，也要处理普通任务和定时任务。整个主循环围绕 select、processSelectedKeys、runAllTasks 三块展开。
1. execute 任务提交入口
外部线程往 EventLoop 提交任务走 SingleThreadEventExecutor#execute。任务先进入 Mpsc 无锁队列；如果是首次提交，startThread 会启动死循环；如果 EventLoop 正在 select 阻塞，还需要 wakeup 唤醒。
public void execute(Runnable task) { if (task == null) { throw new NullPointerException("task"); } boolean inEventLoop = inEventLoop(); addTask(task); if (!inEventLoop) { startThread(); if (isShutdown()) { // 拒绝逻辑 } } if (!addTaskWakesUp && wakesUpForTask(task)) { wakeup(inEventLoop); }}2. wakeup Selector 唤醒
wakeup 只在非 EventLoop 线程调用且 CAS 成功时才真正执行 selector.wakeup。由 EventLoop 自己调用时，效果是取消下一次 select 操作，而不是唤醒当前线程自己。
@Overrideprotected void wakeup(boolean inEventLoop) { if (!inEventLoop && wakenUp.compareAndSet(false, true)) { selector.wakeup(); }}3. doStartThread 主循环启动
startThread 最终走到 doStartThread，把当前线程保存到成员变量 thread，然后调用外部类的 run 方法进入死循环。
private void doStartThread() { assert thread == null; executor.execute(new Runnable() { @Override public void run() { thread = Thread.currentThread(); if (interrupted) { thread.interrupt(); } boolean success = false; updateLastExecutionTime(); try { SingleThreadEventExecutor.this.run(); success = true; } catch (Throwable t) { logger.warn("Unexpected exception from an event executor: ", t); } finally { // 清理工作 } } });}4. NioEventLoop run 循环调度
run 是一个 for 死循环。先通过 selectStrategy.calculateStrategy 决定是直接 selectNow 还是进入阻塞 select；然后 processSelectedKeys 处理 IO 事件；最后按 ioRatio 分配时间执行非 IO 任务。ioRatio 默认 50，表示 IO 与任务各占一半时间。
protected void run() { for (;;) { try { try { switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) { case SelectStrategy.CONTINUE: continue; case SelectStrategy.BUSY_WAIT: case SelectStrategy.SELECT: boolean oldWakenUp = wakenUp.getAndSet(false); select(oldWakenUp); if (wakenUp.get()) { selector.wakeup(); } default: } } catch (IOException e) { rebuildSelector0(); handleLoopException(e); continue; } cancelledKeys = 0; needsToSelectAgain = false; final int ioRatio = this.ioRatio; if (ioRatio == 100) { try { processSelectedKeys(); } finally { runAllTasks(); } } else { final long ioStartTime = System.nanoTime(); try { processSelectedKeys(); } finally { final long ioTime = System.nanoTime() - ioStartTime; runAllTasks(ioTime * (100 - ioRatio) / ioRatio); } } } catch (Throwable t) { handleLoopException(t); } try { if (isShuttingDown()) { closeAll(); if (confirmShutdown()) { return; } } } catch (Throwable t) { handleLoopException(t); } }}5. select 空轮询防治
select 方法内部是一个 for 循环。超时时间由最近定时任务决定，没有定时任务时默认 1 秒。
循环里每醒一次都要判断：有没有 IO 事件、有没有被外部唤醒、有没有新任务、有没有定时任务；任一满足就退出。
关键是 selectCnt 计数。如果 select 在没有真正超时的情况下反复返回，且连续次数超过 SELECTOR_AUTO_REBUILD_THRESHOLD，默认 512，就认为命中了 JDK NIO 空轮询 bug，直接重建 Selector。
private void select(boolean oldWakenUp) throws IOException { Selector selector = this.selector; try { int selectCnt = 0; long currentTimeNanos = System.nanoTime(); long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos); for (;;) { long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L; if (timeoutMillis <= 0) { if (selectCnt == 0) { selector.selectNow(); selectCnt = 1; } break; } if (hasTasks() && wakenUp.compareAndSet(false, true)) { selector.selectNow(); selectCnt = 1; break; } int selectedKeys = selector.select(timeoutMillis); selectCnt++; if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) { break; } if (Thread.interrupted()) { selectCnt = 1; break; } long time = System.nanoTime(); if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) { selectCnt = 1; } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) { selector = selectRebuildSelector(selectCnt); selectCnt = 1; break; } currentTimeNanos = time; } } catch (CancelledKeyException e) { // 记录日志 }}6. processSelectedKey 就绪事件分发
processSelectedKeys 默认把 Selector 内部的 HashSet 替换为数组实现的 SelectedSelectionKeySet，以提升遍历性能。
processSelectedKey 按 readyOps 分发：OP_CONNECT 走 finishConnect，OP_WRITE 走 forceFlush，OP_READ 和 OP_ACCEPT 走 unsafe.read。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe(); if (!k.isValid()) { return; } try { int readyOps = k.readyOps(); if ((readyOps & SelectionKey.OP_CONNECT) != 0) { int ops = k.interestOps(); ops &= ~SelectionKey.OP_CONNECT; k.interestOps(ops); unsafe.finishConnect(); } if ((readyOps & SelectionKey.OP_WRITE) != 0) { ch.unsafe().forceFlush(); } if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { unsafe.read(); } } catch (CancelledKeyException ignored) { unsafe.close(unsafe.voidPromise()); }}三、NioServerSocketChannel Accept 接收流程
boss 线程检测到 OP_ACCEPT 后，走的是 AbstractNioMessageChannel.NioMessageUnsafe#read，而不是 ByteUnsafe 的 read。
它一次可以 accept 多个连接，每个连接封装成 NioSocketChannel 放入 readBuf，再逐个 fireChannelRead。
1. NioMessageUnsafe read 接收循环
doReadMessages 内部执行原生 accept，并把新建立的 SocketChannel 包装成 NioSocketChannel 加入 readBuf。
readBuf 是一个 ArrayList，缓存本次事件里接收到的所有连接。
循环结束后，逐个 fireChannelRead，最终落到 ServerBootstrapAcceptor.channelRead。
public void read() { assert eventLoop().inEventLoop(); final ChannelConfig config = config(); final ChannelPipeline pipeline = pipeline(); final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); allocHandle.reset(config); boolean closed = false; Throwable exception = null; try { try { do { int localRead = doReadMessages(readBuf); if (localRead == 0) { break; } if (localRead < 0) { closed = true; break; } allocHandle.incMessagesRead(localRead); } while (allocHandle.continueReading()); } catch (Throwable t) { exception = t; } int size = readBuf.size(); for (int i = 0; i < size; i++) { readPending = false; pipeline.fireChannelRead(readBuf.get(i)); } readBuf.clear(); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); if (exception != null) { closed = closeOnReadError(exception); pipeline.fireExceptionCaught(exception); } if (closed) { inputShutdown = true; if (isOpen()) { close(voidPromise()); } } } finally { if (!readPending && !config.isAutoRead()) { removeReadOp(); } }}2. ServerBootstrapAcceptor channelRead
channelRead 收到的 msg 就是 NioSocketChannel。它先往 child pipeline 加 childHandler，再设置 childOptions 和 childAttrs，最后调用 childGroup.register 把这个新通道注册到 worker EventLoop。注册失败则 forceClose。
public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; child.pipeline().addLast(childHandler); setChannelOptions(child, childOptions, logger); for (Entry<AttributeKey<?, Object>> e : childAttrs) { child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue()); } try { childGroup.register(child).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { forceClose(child, future.cause()); } } }); } catch (Throwable t) { forceClose(child, t); }}3. NioSocketChannel Register 注册流程
注册逻辑与 boss 侧完全同构：AbstractUnsafe#register 判断当前不在 worker 线程，于是通过 eventLoop.execute 提交任务，完成 boss 到 worker 的线程切换。
register0 里 doRegister 把原生 SocketChannel 绑定到 worker 的 selector，invokeHandlerAddedIfNeeded 执行 childHandler 里的 ChannelInitializer，把用户自定义 handler 加入 pipeline。
doRegister();neverRegistered = false;registered = true;pipeline.invokeHandlerAddedIfNeeded();safeSetSuccess(promise);pipeline.fireChannelRegistered();if (isActive()) { if (firstRegistration) { pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); }}对 NioSocketChannel 而言，accept 成功后通道已经是 active 的，因此注册完成会立刻 fireChannelActive。
4. HeadContext channelActive OP_READ 注册
fireChannelActive 同样传播到 HeadContext，触发 readIfIsAutoRead，进而 doBeginRead。NioSocketChannel 的 readInterestOp 是 OP_READ，于是 worker 线程从此关注该连接的可读事件。
readPending = true;final int interestOps = selectionKey.interestOps();if ((interestOps & readInterestOp) == 0) { selectionKey.interestOps(interestOps | readInterestOp);}四、NioSocketChannel Read 读取流程
worker 线程检测到 OP_READ 后，走 AbstractNioByteChannel.NioByteUnsafe#read。一次 IO 事件可能读不完所有数据，因此会触发多次 nio read。
单次 nio read 事件内部又会按 continueReading 循环多次分配 ByteBuf 并 fireChannelRead，最后统一 fireChannelReadComplete。
1. NioByteUnsafe read 读取循环
allocHandle.allocate 根据上次读取经验估算本次 ByteBuf 大小；doReadBytes 把数据写进 ByteBuf；lastBytesRead 小于 0 表示对端关闭，等于 0 表示暂时没数据。每次读到数据就 fireChannelRead 给 pipeline，由业务 handler 处理。
public final void read() { final ChannelConfig config = config(); if (shouldBreakReadReady(config)) { clearReadPending(); return; } final ChannelPipeline pipeline = pipeline(); final ByteBufAllocator allocator = config.getAllocator(); final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle(); allocHandle.reset(config); ByteBuf byteBuf = null; boolean close = false; try { do { byteBuf = allocHandle.allocate(allocator); allocHandle.lastBytesRead(doReadBytes(byteBuf)); if (allocHandle.lastBytesRead() <= 0) { byteBuf.release(); byteBuf = null; close = allocHandle.lastBytesRead() < 0; if (close) { readPending = false; } break; } allocHandle.incMessagesRead(1); readPending = false; pipeline.fireChannelRead(byteBuf); byteBuf = null; } while (allocHandle.continueReading()); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); if (close) { closeOnRead(pipeline); } } catch (Throwable t) { handleReadException(pipeline, byteBuf, t, close, allocHandle); } finally { if (!readPending && !config.isAutoRead()) { removeReadOp(); } }}2. continueReading 循环控制
continueReading 决定一次 IO 事件里要不要继续读。四个条件同时成立才继续：
autoRead 为 true
respectMaybeMoreData 为 true 且实际读取字节等于预估值，说明缓冲区可能还没读完
累计消息数小于 maxMessagePerRead，默认 16
累计字节数大于 0
public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) { return config.isAutoRead() && (!respectMaybeMoreData || maybeMoreDataSupplier.get()) && totalMessages < maxMessagePerRead && totalBytesRead > 0;}这套机制让 Netty 在一次 select 唤醒后尽量把可读数据读干净，同时通过 maxMessagePerRead 限制单次循环次数，避免长时间霸占 worker 线程而饿死其他通道和任务。
文章结束，喜欢就给个关注吧，你的肯定是我最大的动力，点赞上一千我就是脑瘫也出下章。
