目录

类关系图4 bind4.1 initAndRegister4.1.1 newChannel4.1.2 init4.1.2.1 ServerBootstrapAcceptor 4.1.3 register4.1.3.1 next4.1.3.2 register 4.2 doBind04.2.1 doBind4.2.2 fireChannelActive

类关系图

4 bind

创建NioServerSocketChannel、pipeline、首尾handler、ServerBootstrapAcceptor 、绑定 Selector 调用 HeadContext 的 bind 和 read，绑定端口，注册接收连接事件

类 AbstractBootstrap

public ChannelFuture bind() { validate(); SocketAddress localAddress = this.localAddress; if (localAddress == null) { throw new IllegalStateException("localAddress not set"); } return doBind(localAddress); }

跟 doBind：

private ChannelFuture doBind(final SocketAddress localAddress) { //创建NioServerSocketChannel、pipeline、首尾handler、ServerBootstrapAcceptor 见4.1 final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; } if (regFuture.isDone()) { // At this point we know that the registration was complete and successful. ChannelPromise promise = channel.newPromise(); //调用 HeadContext 的 bind 和 read，绑定端口，注册接收连接事件 见4.2 doBind0(regFuture, channel, localAddress, promise); return promise; } else { // Registration future is almost always fulfilled already, but just in case it's not. 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) { // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an // IllegalStateException once we try to access the EventLoop of the Channel. promise.setFailure(cause); } else { // Registration was successful, so set the correct executor to use. // See https://github.com/netty/netty/issues/2586 promise.registered(); doBind0(regFuture, channel, localAddress, promise); } } }); return promise; } }

4.1 initAndRegister

4.1.1 创建NioServerSocketChannel、pipeline、首尾handler 4.1.2 增加一个 ServerBootstrapAcceptor 在尾 handler 前

类 AbstractBootstrap

final ChannelFuture initAndRegister() { Channel channel = null; try { //创建NioServerSocketChannel、pipeline、首尾handler 见4.1.1 channel = channelFactory.newChannel(); //增加一个 ServerBootstrapAcceptor 在尾handler前 见4.1.2 init(channel); } catch (Throwable t) { if (channel != null) { // channel can be null if newChannel crashed (eg SocketException("too many open files")) channel.unsafe().closeForcibly(); // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t); } //见4.1.3 ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }

4.1.1 newChannel

创建NioServerSocketChannel、pipeline、首尾handler

类 AbstractBootstrap

@Override public T newChannel() { try { //通过无参构造反射 return clazz.getConstructor().newInstance(); } catch (Throwable t) { throw new ChannelException("Unable to create Channel from class " + clazz, t); } }

这里创建的是 NioServerSocketChannel

将成员变量 readInterestOp 设置为 SelectionKey.OP_ACCEPT，但还没有真正的绑定事件

4.1.2 init

增加一个 ServerBootstrapAcceptor 在尾 handler 前

类 ServerBootStrap

@Override 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) { //自己写的handler，通常没有。自己创建的一般是childHandler处理读写。 pipeline.addLast(handler); } //异步 见 3.3 ch.eventLoop().execute(new Runnable() { @Override public void run() { //增加一个ServerBootstrapAcceptor 见 4.1.2.1 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }

4.1.2.1 ServerBootstrapAcceptor

ServerBootStrap 的内部类，是一个 ChannelInboundHandler

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter { private final EventLoopGroup childGroup; private final ChannelHandler childHandler; private final Entry<ChannelOption<?>, Object>[] childOptions; private final Entry<AttributeKey<?>, Object>[] childAttrs; private final Runnable enableAutoReadTask; ServerBootstrapAcceptor( final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler, Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) { this.childGroup = childGroup; this.childHandler = childHandler; this.childOptions = childOptions; this.childAttrs = childAttrs; // Task which is scheduled to re-enable auto-read. // It's important to create this Runnable before we try to submit it as otherwise the URLClassLoader may // not be able to load the class because of the file limit it already reached. // // See https://github.com/netty/netty/issues/1328 enableAutoReadTask = new Runnable() { @Override public void run() { channel.config().setAutoRead(true); } }; } @Override @SuppressWarnings("unchecked") public void channelRead(ChannelHandlerContext ctx, Object msg) { //传给accepter的msg是channel（netty的） 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); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { final ChannelConfig config = ctx.channel().config(); if (config.isAutoRead()) { // stop accept new connections for 1 second to allow the channel to recover // See https://github.com/netty/netty/issues/1328 config.setAutoRead(false); ctx.channel().eventLoop().schedule(enableAutoReadTask, 1, TimeUnit.SECONDS); } // still let the exceptionCaught event flow through the pipeline to give the user // a chance to do something with it ctx.fireExceptionCaught(cause); } //...

4.1.3 register

绑定 Selector

类 MultithreadEventLoopGroup

@Override public ChannelFuture register(Channel channel) { //4.1.3.1 //4.1.3.2 return next().register(channel); }

4.1.3.1 next

类 MultithreadEventLoopGroup

@Override public EventLoop next() { return (EventLoop) super.next(); }

跟 next： 类 MultithreadEventExecutorGroup

@Override public EventExecutor next() { return chooser.next(); }

跟 next： 类 GenericEventExecutorChooser 是 DefaultEventExecutorChooserFactory 的一个内部类

从 EventLoopGroup 里轮询一个 EventLoop

@Override public EventExecutor next() { return executors[Math.abs(idx.getAndIncrement() % executors.length)]; }

4.1.3.2 register

类 SingleThreadEventLoop

@Override public ChannelFuture register(Channel channel) { return register(new DefaultChannelPromise(channel, this)); }

跟 register：

@Override public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise"); promise.channel().unsafe().register(this, promise); return promise; }

跟 register：

@Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } if (isRegistered()) { promise.setFailure(new IllegalStateException("registered to an event loop already")); return; } if (!isCompatible(eventLoop)) { promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; } AbstractChannel.this.eventLoop = eventLoop; //如果当前线程就是EventLoop里的成员变量 if (eventLoop.inEventLoop()) { register0(promise); } else { try { //见3.3 eventLoop.execute(new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { logger.warn( "Force-closing a channel whose registration task was not accepted by an event loop: {}", AbstractChannel.this, t); closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } } }

跟 register0： 类 AbstractChannel

private void register0(ChannelPromise promise) { try { // check if the channel is still open as it could be closed in the mean time when the register // call was outside of the eventLoop if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; doRegister(); neverRegistered = false; registered = true; // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the // user may already fire events through the pipeline in the ChannelFutureListener. pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); pipeline.fireChannelRegistered(); // Only fire a channelActive if the channel has never been registered. This prevents firing // multiple channel actives if the channel is deregistered and re-registered. if (isActive()) { if (firstRegistration) { pipeline.fireChannelActive(); } else if (config().isAutoRead()) { // This channel was registered before and autoRead() is set. This means we need to begin read // again so that we process inbound data. // // See https://github.com/netty/netty/issues/4805 beginRead(); } } } catch (Throwable t) { // Close the channel directly to avoid FD leak. closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } }

跟 doRegister：

@Override protected void doRegister() throws Exception { boolean selected = false; for (;;) { try { //jdk里的channel 0，没有注册事件 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this); return; } catch (CancelledKeyException e) { if (!selected) { // Force the Selector to select now as the "canceled" SelectionKey may still be // cached and not removed because no Select.select(..) operation was called yet. eventLoop().selectNow(); selected = true; } else { // We forced a select operation on the selector before but the SelectionKey is still cached // for whatever reason. JDK bug ? throw e; } } } }

4.2 doBind0

调用 HeadContext 的 bind 和 read，绑定端口，注册接收连接事件

类 AbstractBootstrap

private static void doBind0( final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) { // This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up // the pipeline in its channelRegistered() implementation. 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()); } } }); }

跟 bind： 类 AbstractChannel

@Override public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) { return pipeline.bind(localAddress, promise); }

跟 bind： 类 DefaultChannelPipeline

绑定到 tail 证明是出站事件，因为出站事件第一个遇到的才是 tail。如果是入站事件，应该绑定 head。 然而，tail 是 ChannelInboundHandler，入站处理器

@Override public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) { return tail.bind(localAddress, promise); }

跟 bind： 类 AbstractChannelHandlerContext

从尾往前找到第一个出站处理器，找到的是首handler。 前面提到过了，服务器一般不会自定义 handler。自定义的是 childHandler，处理读写。handler 是处理连接的。 客户端才会定义 handler，处理读写。 可以看一下这个示例

@Override public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) { if (localAddress == null) { throw new NullPointerException("localAddress"); } if (isNotValidPromise(promise, false)) { // cancelled return promise; } //找出站处理器，从尾往前找到第一个出站处理器 final AbstractChannelHandlerContext next = findContextOutbound(); EventExecutor executor = next.executor(); if (executor.inEventLoop()) { next.invokeBind(localAddress, promise); } else { safeExecute(executor, new Runnable() { @Override public void run() { next.invokeBind(localAddress, promise); } }, promise, null); } return promise; }

跟 invokeBind：

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) { if (invokeHandler()) { try { ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise); } catch (Throwable t) { notifyOutboundHandlerException(t, promise); } } else { bind(localAddress, promise); } }

跟 bind： 类 HeadContext，是 DefaultChannelPipeline 的内部类

@Override public void bind( ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception { unsafe.bind(localAddress, promise); }

跟 bind： 类 AbstractUnsafe，是 AbstractChannel 的内部类

@Override public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { assertEventLoop(); if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } // See: https://github.com/netty/netty/issues/576 if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) && localAddress instanceof InetSocketAddress && !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() && !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) { // Warn a user about the fact that a non-root user can't receive a // broadcast packet on *nix if the socket is bound on non-wildcard address. logger.warn( "A non-root user can't receive a broadcast packet if the socket " + "is not bound to a wildcard address; binding to a non-wildcard " + "address (" + localAddress + ") anyway as requested."); } boolean wasActive = isActive(); try { //4.2.1 doBind(localAddress); } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { invokeLater(new Runnable() { @Override public void run() { //入站事件 见4.2.2 pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); }

4.2.1 doBind

类 NioServerSocketChannel

@Override protected void doBind(SocketAddress localAddress) throws Exception { if (PlatformDependent.javaVersion() >= 7) { //jdk的bind javaChannel().bind(localAddress, config.getBacklog()); } else { javaChannel().socket().bind(localAddress, config.getBacklog()); } }

4.2.2 fireChannelActive

类 DefaultChannelPipeline

从首 handler 开始调用所有入站处理器的 channelActive 方法

@Override public final ChannelPipeline fireChannelActive() { AbstractChannelHandlerContext.invokeChannelActive(head); return this; }

跟 invokeChannelActive： 类 AbstractChannelHandlerContext

static void invokeChannelActive(final AbstractChannelHandlerContext next) { EventExecutor executor = next.executor(); if (executor.inEventLoop()) { //head next.invokeChannelActive(); } else { executor.execute(new Runnable() { @Override public void run() { next.invokeChannelActive(); } }); } }

跟 invokeChannelActive：

private void invokeChannelActive() { if (invokeHandler()) { try { ((ChannelInboundHandler) handler()).channelActive(this); } catch (Throwable t) { notifyHandlerException(t); } } else { fireChannelActive(); } }

跟 channelActive： 类 HeadContext，DefaultChannelPipeline 的内部类

@Override public void channelActive(ChannelHandlerContext ctx) throws Exception { //调用后面入站handler的channelActive方法。 //因为ServerBootstrapAcceptor和TailContext的channelActive都没有操作，所以看下面的方法就够了。 ctx.fireChannelActive(); readIfIsAutoRead(); }

跟 readIfIsAutoRead：

private void readIfIsAutoRead() { if (channel.config().isAutoRead()) { channel.read(); } }

跟 read： 类 AbstractChannel

@Override public Channel read() { pipeline.read(); return this; }

跟 read：

@Override public final ChannelPipeline read() { //出站事件，由出站handler处理，回去看HeadContext的这个方法 tail.read(); return this; }

类 HeadContext

@Override public void read(ChannelHandlerContext ctx) { unsafe.beginRead(); }

跟 beginRead： 类 AbstractUnsafe in AbstractChannel

@Override public final void beginRead() { assertEventLoop(); if (!isActive()) { return; } try { doBeginRead(); } catch (final Exception e) { invokeLater(new Runnable() { @Override public void run() { pipeline.fireExceptionCaught(e); } }); close(voidPromise()); } }

跟 doBeginRead： 类 AbstractNioChannel

@Override protected void doBeginRead() throws Exception { // Channel.read() or ChannelHandlerContext.read() was called final SelectionKey selectionKey = this.selectionKey; if (!selectionKey.isValid()) { return; } readPending = true; final int interestOps = selectionKey.interestOps(); if ((interestOps & readInterestOp) == 0) { //注册SelectionKey.OP_ACCEPT 见（5） selectionKey.interestOps(interestOps | readInterestOp); } }