Netty源码分析-服务器端启动流程

参考文献

• 黑马Netty

环境

• 基于Netty4.1.92.Final-SNAPSHOT

启动流程

• Netty中启动流程大致处理步骤

  //1 netty 中使用 NioEventLoopGroup （简称 nio boss 线程）来封装线程和 selector
  Selector selector = Selector.open(); 
  
  //2 创建 NioServerSocketChannel,同时会初始化它关联的 handler,以及为原生 ssc 存储 config
  NioServerSocketChannel attachment = new NioServerSocketChannel();
  
  //3 创建 NioServerSocketChannel 时,创建了 java 原生的 ServerSocketChannel
  ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 
  serverSocketChannel.configureBlocking(false);
  
  //4 启动 nio boss 线程执行接下来的操作
  
  //5 注册（仅关联 selector 和 NioServerSocketChannel）,未关注事件
  SelectionKey selectionKey = serverSocketChannel.register(selector, 0, attachment);
  
  //6 head -> 初始化器 -> ServerBootstrapAcceptor -> tail,初始化器是一次性的,只为添加 acceptor
  
  //7 绑定端口
  serverSocketChannel.bind(new InetSocketAddress(8080));
  
  //8 触发 channel active 事件,在 head 中关注 op_accept 事件
  selectionKey.interestOps(SelectionKey.OP_ACCEPT);

• 创建服务器端示例

  final NioEventLoopGroup boosGroup = new NioEventLoopGroup(MAIN_THREAD_NUMBER);
      final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
      final ServerBootstrap serverBootstrap = new ServerBootstrap();
      try {
          serverBootstrap.group(boosGroup, workerGroup)
                  .channel(NioServerSocketChannel.class)
                  .handler(new LoggingHandler("DEBUG"))
                  .childHandler(initializer);
          final ChannelFuture channelFuture = serverBootstrap.bind(port);
          channelFuture.addListener(listener -> {
              if (listener.isSuccess()) {
                  LOGGER.info("WebSocket绑定端口:{}成功", port);
              } else {
                  LOGGER.warn("WebSocket绑定端口:{}失败", port);
              }
          });
          final Channel channel = channelFuture.sync().channel();
          channel.closeFuture().sync();
      } finally {
          boosGroup.shutdownGracefully();
          workerGroup.shutdownGracefully();
      }

入口

• io.netty.bootstrap.AbstractBootstrap#bind(java.net.SocketAddress)
• 所处线程: main

/**
   * Create a new {@link Channel} and bind it.
   */
  public ChannelFuture bind(SocketAddress localAddress) {
      validate();
      return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
  }

private ChannelFuture doBind(final SocketAddress localAddress) {
        // 1. 执行初始化和注册 regFuture 会由 initAndRegister 设置其是否完成,从而回调 3.2 处代码
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        // 2. 因为是 initAndRegister 异步执行,需要分两种情况来看,调试时也需要通过 suspend 断点类型加以区分
        // 2.1 如果已经完成
        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            // 3.1 立刻调用 doBind0
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // 2.2 还没有完成
            // Registration future is almost always fulfilled already, but just in case it's not.
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            // 3.2 通过回调来执行doBind0
            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();
                        // 3. 由注册线程去执行 doBind0
                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 1.0 此处创建Channel方法一般为io.netty.channel.ReflectiveChannelFactory.newChannel(),通过反射来创建NioServerSocketChannel
        channel = channelFactory.newChannel();
        // 1.1 初始化 - 做的事就是添加一个初始化器 ChannelInitializer
        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);
    }
    
    // 1.2 注册 - 做的事就是将原生 channel 注册到 selector 上
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }

    // If we are here and the promise is not failed, it's one of the following cases:
    // 1) If we attempted registration from the event loop, the registration has been completed at this point.
    //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
    // 2) If we attempted registration from the other thread, the registration request has been successfully
    //    added to the event loop's task queue for later execution.
    //    i.e. It's safe to attempt bind() or connect() now:
    //         because bind() or connect() will be executed *after* the scheduled registration task is executed
    //         because register(), bind(), and connect() are all bound to the same thread.

    return regFuture;
}

创建NioServerSocketChannel

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 1.0 此处创建Channel方法一般为io.netty.channel.ReflectiveChannelFactory.newChannel(),通过反射来创建NioServerSocketChannel
        channel = channelFactory.newChannel();
    } catch (Throwable t) {
    }
}

• 执行后会进入NioServerSocketChannel的默认构造方法

/**
 * Create a new instance
 */
public NioServerSocketChannel() {
    this(DEFAULT_SELECTOR_PROVIDER);
}
/**
 * Create a new instance using the given {@link SelectorProvider}.
 */
public NioServerSocketChannel(SelectorProvider provider) {
    this(provider, null);
}

/**
 * Create a new instance using the given {@link SelectorProvider} and protocol family (supported only since JDK 15).
 */
public NioServerSocketChannel(SelectorProvider provider, InternetProtocolFamily family) {
    this(newChannel(provider, family));
}

/**
 * Create a new instance using the given {@link ServerSocketChannel}.
 */
public NioServerSocketChannel(ServerSocketChannel channel) {
    // 对于服务端来说,关心的是 SelectionKey.OP_ACCEPT 事件,等待客户端连接
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

• 使用默认构造函数会依次调用上面三个带参数的构造函数
• newChannel(provider, family)最终使用provider.openServerSocketChannel()来创建ServerSocketChannel

private static final Method OPEN_SERVER_SOCKET_CHANNEL_WITH_FAMILY =
          SelectorProviderUtil.findOpenMethod("openServerSocketChannel");

  private static ServerSocketChannel newChannel(SelectorProvider provider, InternetProtocolFamily family) {
      try {
          ServerSocketChannel channel =
                  SelectorProviderUtil.newChannel(OPEN_SERVER_SOCKET_CHANNEL_WITH_FAMILY, provider, family);
          return channel == null ? provider.openServerSocketChannel() : channel;
      } catch (IOException e) {
          throw new ChannelException("Failed to open a socket.", e);
      }
  }

此处的ChannelFactory为什么会是io.netty.channel.ReflectiveChannelFactory

• 因为在示例中,我们通过设置.channel(NioServerSocketChannel.class)指定了ChannelFactory为io.netty.channel.ReflectiveChannelFactory

  serverBootstrap.group(boosGroup, workerGroup)
                   .channel(NioServerSocketChannel.class)
                   .handler(new LoggingHandler("DEBUG"))
                   .childHandler(initializer);

  public B channel(Class<? extends C> channelClass) {
      return channelFactory(new ReflectiveChannelFactory<C>(
              ObjectUtil.checkNotNull(channelClass, "channelClass")
      ));
  }

初始化Channel

• 以创建NioServerSocketChannel为例
• io.netty.bootstrap.ServerBootstrap#init
• 所处线程: main

@Override
void init(Channel channel) {
    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);

    // 为 NioServerSocketChannel 添加初始化器
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) {
          	// 此处的代码会在NIO线程中执行
          	// 具体由io.netty.channel.AbstractChannel.AbstractUnsafe#register0中的
          	// pipeline.invokeHandlerAddedIfNeeded();触发
            final ChannelPipeline pipeline = ch.pipeline();
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }
            // 初始化器的职责是将 ServerBootstrapAcceptor 加入至 NioServerSocketChannel
          	// ServerBootstrapAcceptor的作用是在accrpt事件发生后建立连接
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}

• pipeline中的handler为head-->ServerBootstrapAcceptor-->tail

注册

• 调用链

io.netty.channel.EventLoopGroup#register(io.netty.channel.Channel)
	-->io.netty.channel.SingleThreadEventLoop#register(io.netty.channel.Channel)
		-->io.netty.channel.SingleThreadEventLoop#register(io.netty.channel.ChannelPromise)
			-->io.netty.channel.Channel.Unsafe#register
				-->io.netty.channel.AbstractChannel.AbstractUnsafe#register
					-->io.netty.channel.AbstractChannel.AbstractUnsafe#register0
						-->io.netty.channel.nio.AbstractNioChannel#doRegister

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    ObjectUtil.checkNotNull(eventLoop, "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为BossGroup里面的线程,此时执行代码的线程为Main线程,因此eventLoop.inEventLoop()为false
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            // 首次执行 execute 方法时,会启动 nio 线程,之后注册等操作在 nio 线程上执行
            // 因为只有一个 NioServerSocketChannel 因此,也只会有一个 boss nio 线程
            // 这行代码完成的事实是 main -> nio boss 线程的切换
            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所属线程为NIO线程

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;
        // 1.2.1 原生的 nio channel 绑定到 selector 上,注意此时没有注册 selector 关注事件,附件为 NioServerSocketChannel
        doRegister();
        neverRegistered = false;
        registered = true;

        // 1.2.2 执行 NioServerSocketChannel 初始化器ChannelInitializer的 initChannel
        // 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();

        // 回调 3.2 io.netty.bootstrap.AbstractBootstrap#doBind0
        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        // 对应 server socket channel 还未绑定,isActive 为 false
        // 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);
    }
}

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // javaChannel()获取的就是java.nio.channels.ServerSocketChannel
            // eventLoop().unwrappedSelector()获取的就是NioEventLoop中java.nio.channels.spi.SelectorProvider.openSelector()返回的Selector
            // this就是之前通过反射创建的NioServerSocketChannel,作为附件注册到java.nio.channels.ServerSocketChannel上
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
          	// 此时Channel没有关注任何事件
            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;
            }
        }
    }
}

绑定端口

• 入口: io.netty.bootstrap.AbstractBootstrap#doBind0

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.
  	// 切换到NIO线程
    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的调用链

io.netty.channel.ChannelOutboundInvoker#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
-->io.netty.channel.AbstractChannel#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
	-->io.netty.channel.DefaultChannelPipeline#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
		-->io.netty.channel.AbstractChannelHandlerContext#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
			-->io.netty.channel.AbstractChannelHandlerContext#invokeBind
				-->io.netty.channel.AbstractChannel.AbstractUnsafe#bind
					-->io.netty.channel.socket.nio.NioServerSocketChannel#doBind

@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 {
        // 3.3 执行端口绑定
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }

    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
              	// 3.4 触发 active 事件
                pipeline.fireChannelActive();
            }
        });
    }

    safeSetSuccess(promise);
}

执行端口绑定

@SuppressJava6Requirement(reason = "Usage guarded by java version check")
@Override
protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}

触发 active 事件

• 调用链

  io.netty.channel.DefaultChannelPipeline#fireChannelActive
  -->io.netty.channel.AbstractChannelHandlerContext#invokeChannelActive(io.netty.channel.AbstractChannelHandlerContext)
  	-->io.netty.channel.AbstractChannelHandlerContext#invokeChannelActive()
  		-->io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive
    		-->io.netty.channel.DefaultChannelPipeline.HeadContext#readIfIsAutoRead
    				-->io.netty.channel.AbstractChannel#read
    					-->io.netty.channel.DefaultChannelPipeline#read
    						-->io.netty.channel.AbstractChannelHandlerContext#read
    							-->io.netty.channel.AbstractChannelHandlerContext#invokeRead
    								-->io.netty.channel.DefaultChannelPipeline.HeadContext#read
    									-->io.netty.channel.AbstractChannel.AbstractUnsafe#beginRead
    										-->io.netty.channel.nio.AbstractNioChannel#doBeginRead

@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();
    // readInterestOp 取值是 16,在 NioServerSocketChannel 创建时初始化好,代表关注 accept 事件
  	// 在 NioServerSocketChannel创建时调用父类AbstractNioChannel的构造方法时设置了Channel的非阻塞模式
  	// ch.configureBlocking(false);
    if ((interestOps & readInterestOp) == 0) {
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}