[Netty源码] 服务端启动过程

Posted 959y

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了[Netty源码] 服务端启动过程相关的知识，希望对你有一定的参考价值。

文章目录

1.ServerBootstrap

ServerBootstrap引导服务端启动流程:


//主EventLoopGroup
NioEventLoopGroup master = new NioEventLoopGroup();
//从EventLoopGroup
NioEventLoopGroup worker = new NioEventLoopGroup();
//服务端引导类
ServerBootstrap bootstrap = new ServerBootstrap();
//配置主从EventLoopGroup
bootstrap.group(master, worker);
//channel选项配置
bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
  .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000);
bootstrap.channel(NioserverSocketChannel.class);
//主EventLoopGroup ChannelHandler配置
bootstrap.handler(new ChannelInboundHandlerAdapter()
  @Override
  public void channelRegistered(ChannelHandlerContext ctx) throws Exception 
    System.out.println("Master:" + ctx.channel().eventLoop().toString());
    super.channelRegistered(ctx);
  
);
//从EventLoopGroup ChannelHandler配置
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() 
  @Override
  protected void initChannel(SocketChannel ch) throws Exception 
    System.out.println("Child:" + ch.eventLoop().toString());
  
);
//调用bind方法开始监听端口8888
ChannelFuture future = bootstrap.bind(8888).sync();
future.channel().closeFuture().sync();

master.shutdownGracefully().sync();

以上代码使用的是Netty框架中经典的主从事件驱动模式

2.服务端启动过程

  1. main方法调用ServerBootstrap.bind()方法

  2. validate()方法验证必要配置参数

  3. 调用AbstractBootstrap.initAndRegister()方法

  4. 调用channelFactory.newChannel()方法创建Channel实例

  5. ServerBootstrap.init()初始化Channel实例，配置ChannelOption、attributes；在Channel的pipeline中配置默认的ChannelHandler实例

  6. 将Channel实例注册到主EventLoopGroup中并返回ChannelFuture

  7. 注册ChannelFuture回调，在完成后调用channel.bind()方法完成端口监听

3.具体步骤分析

1.创建服务端Channel
2.初始化服务端Channel
3.注册Selector
4.端口绑定
5.触发读事件

3.1 创建服务端Channel

AbstractBootstrap.bind()

  1. validate(): 验证必要配置参数
  2. AbstractBootstrap.doBind()

AbstractBootstrap.initAndRegister()

  1. 利用反射创建Channel, 得到NioServerSocketChannel
  2. 初始化channel中的一些参数

AbstractBootstrap.channel() 反射创建Channel

利用反射创建channel, 得到ReflectiveChannelFactory, 调用channelFactory利用工厂模式, 最后生成NioServerSocketChannel

3.2 初始化服务端Channel

ServerBootstrap.init() 初始化一些基本参数

初始化一些Options和Attrs和group和handler的参数

    @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) 
                    pipeline.addLast(handler);
                

                ch.eventLoop().execute(new Runnable() 
                    @Override
                    public void run() 
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    
                );
            
        );

设置 currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs

在Channel的pipeline中配置默认的ChannelHandler实例

ServerBootstrapAcceptor 添加连接器, ServerBootstrapAcceptor本质是一个handler

3.3 注册selector

bind -> initAndRegister -> AbstractChannel.register -> this.eventLoop = eventLoop, register0 实际注册 -> doRegister(), invokeHandlerAddedIfNeeded(), fireChannelRegistered() 传播事件

AbstractChannel.AbstractUnsafe.register()

register0()

AbstractNioChannel.doRegister()

上面有javaChannel.register(), 利用jdk底层去创建selector

3.4 端口绑定

bind() -> doBind() -> doBind0() -> AbstracChannel,bind() -> DefaultChannelPipeline.bind() -> AbstractChannlHandlerContext.bind() -> AbstractChannlHandlerContext.invokeBind() -> DefaultChannelPipeline.HeadContext.bind() -> NioSocketChannel.doBind0(), pipeline.fireChannelActive();

最后将会调用DefaultChannelPipeline.HeadContext.bind(), 因为DefaultChannelPipeline在初始化时会设置pipeline队列的首尾分别为DefaultChannelPipeline.HeadContext与DefaultChannelPipeline.TailContext
bind()在Pipeline中走的是出站方法，是从管道的后面向前走，最后到达管道头部的ChannelHandler(也就是DefaultChannelPipeline.HeadContext)，这一过程会调用同一方向上所有ChannelHandler的bind()事件。

AbstractBootstrap.doBind0(): 添加一个任务至EventLoop中, 最后将会调用DefaultChannelPipeline.HeadContext.bind()方法

AbstractChannlHandlerContext.invokeBind()

DefaultChannelPipeline.HeadContext.bind() -> AbstractChannel.AbstractUnsafe.bind()

        @Override
        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);
  1. NioSocketChannel.doBind() -> NioSocketChannel.doBind0()
  2. pipeline.fireChannelActive()

通过在pipline中的层层传递, 现在来到了最终执行bind操作的终点, 执行bind方法,

可以通过SocketUtils.bind() 绑定JDK底层的端口

3.5 服务端的读事件

最后如果触发了一个事件的话, 会调用Channel.read()事件(AbstractNioChannel.doBeginRead()), 这个事件对于服务端来说就是一个新的连接接入。

AbstractChannel.AbstractUnsafe.bind()

HeadContext.channelActive()

这个tail.read(), 触发的是beginRead()方法

AbstractUnsafe.beginRead()

AbstractNioChannel.doBeginRead()

这里的readInterestOp就是Acept事件

服务端端口绑定成功, 触发一个Acept事件, 然后调用channel.read()事件, 对于服务端来说就相当于可以读了, 可以读取一个新的连接。

图说Netty服务端启动过程

我们知道Netty是一个基于JDK的nio实现的网络编程框架，那Netty的服务端是怎么启动的呢，包括他是何时register 的，何时 bind 端口的，以及何时开始读取网络中的数据的？

让我们带着这个疑问，通过一个官方的例子来深入探究Netty服务端的启动过程。

PS:本文基于netty源码的4.1分支进行分析。

首先我们拿一个最简单的EchoServer的例子来举例说明，具体的代码如下：

 
   
   
 
    
  
    
    
  
  1. EventLoopGroup bossGroup = new NioEventLoopGroup(1);  // 1
    2. 
  
    
    
  
  2. EventLoopGroup workerGroup = new NioEventLoopGroup();
    3. 
  
    
    
  
  3. try {
    4. 
  
    
    
  
  4.     ServerBootstrap b = new ServerBootstrap();
    5. 
  
    
    
  
  5.     b.group(bossGroup, workerGroup) 
    6. 
  
    
    
  
  6.      .channel(NioServerSocketChannel.class) // 2
    7. 
  
    
    
  
  7.      .option(ChannelOption.SO_BACKLOG, 100)
    8. 
  
    
    
  
  8.      .handler(new LoggingHandler(LogLevel.INFO)) // 3
    9. 
  
    
    
  
  9.      .childHandler(new ChannelInitializer<SocketChannel>() {  // 4
    10. 
  
    
    
  
  10.          @Override
    11. 
  
    
    
  
  11.          public void initChannel(SocketChannel ch) throws Exception {
    12. 
  
    
    
  
  12.              ChannelPipeline p = ch.pipeline();
    13. 
  
    
    
  
  13.              if (sslCtx != null) {
    14. 
  
    
    
  
  14.                  p.addLast(sslCtx.newHandler(ch.alloc()));
    15. 
  
    
    
  
  15.              }
    16. 
  
    
    
  
  16.              p.addLast(new EchoServerHandler());
    17. 
  
    
    
  
  17.          }
    18. 
  
    
    
  
  18.      });
    19. 
  
    
    
  
    20. 
  
    
    
  
  20.     // Start the server.
    21. 
  
    
    
  
  21.     ChannelFuture f = b.bind(PORT).sync(); // 5
    22. 
  
    
    
  
    23. 
  
    
    
  
  23.     // Wait until the server socket is closed.
    24. 
  
    
    
  
  24.     f.channel().closeFuture().sync();
    25. 
  
    
    
  
  25. } finally {
    26. 
  
    
    
  
  26.     // Shut down all event loops to terminate all threads.
    27. 
  
    
    
  
  27.     bossGroup.shutdownGracefully();
    28. 
  
    
    
  
  28.     workerGroup.shutdownGracefully();
    29. 
  
    
    
  
  29. }
    30.

从上面的代码来看，在启动的过程中共有5处地方需要我们关注，不过最重要的启动服务端的代码，还是在最后第5步的时候。

为了更加清晰的描述整个启动的过程，也便于我们更好的理解和记忆，我将使用多图形少代码的形式来表达。

首先我把启动过程的一个大致流程画成如下的图：

其中有以下几个核心的方法：

  • channel()

  • handler()

  • childHandler()

  • doBind()

除此之外，还有一个初始化EventLoopGroup类的方法：

  • NioEventLoopGroup()

一、初始化EventLoopGroup

我们从最初的初始化 EventLoopGroup 类开始吧，从源码中可以看到是一层一层的构造方法的调用，然后再super到了父类中，最终会调用到 AbstractEventExecutor 类，具体的调用流程如下图所示：

图说Netty服务端启动过程

这个过程中创建了几个重要的实例，我用淡蓝色标记出来了。

首先我们需要知道的是，在Netty中有几个比较重要的类：

  • EventLoop

  • EventLoopGroup

  • EventExecutor

  • EventExecutorGroup

他们之间的关系图如下所示：

图说Netty服务端启动过程

EventLoop和EventExecutor说到底都是一种Executor。

然后通过调用ServerBootstrap的group()方法，我们将创建的EventLoopGroup对象分别赋值给了ServerBootstrap的 group 和 childGroup 属性。

二、执行channel()方法

初始化完了EventLoopGroup之后，接着就开始执行 channel() 方法了，这个方法很简单，就是通过 ReflectiveChannelFactory 类创建了一个 channelFactory ，这个 channelFactory 后面会很有用，都是通过它来创建需要的Channel实例的。这里我就不贴具体的代码了，具体的执行过程可以用下面的图来表示：

图说Netty服务端启动过程

通过调用该方法，ServerBootstrap类的 channelFactory 属性就被赋予了值，并且该ChannelFactory的实现类是通过反射来创建Channel的。

后面在需要创建Channel的时候，会调用该channelFactory的 newChannel() 方法，执行该方法之后，会创建三种非常有用的对象：

  • channel

  • pipeline

  • unsafe

三、执行handler()方法

该方法没有创建其他的对象，只是把用户提供的方法参数中所表示的ChannelHandler对象通过该方法来赋值给ServerBootstrap的 handler 属性。

PS:这里创建的handler在后面的初始化时会使用到

四、执行childHandler()方法

该方法没有创建其他的对象，只是把用户提供的方法参数中所表示的ChannelHandler对象通过该方法来赋值给ServerBootstrap的 childHandler 属性。

PS:这里创建的childHandler在后面的初始化时会使用到

五、执行doBind()方法

Netty启动过程中最复杂，步骤最多的就是这个方法了，不过不用担心，我已经把该方法核心的执行过程整理好了，如下图所示：

图说Netty服务端启动过程

这里我推荐大家在读源码的时候，可以拿一张纸，一支笔，用画图的形式把方法的调用过程，以及创建了哪些属性等等这些都记下来，一开始可以不用知道那些方法和属性具体是干什么的。先把整个调用流程理清楚，然后再一点一点细化，由点到面的扩展开来，最终把你那张图丰富成一个完整的调用图。

从图中可以看的出来，doBind方法拆分成了两个核心的方法：

  • initAndRegister()

  • doBind0()

第一个 initAndRegister 方法，从方法名字上就可以看得出来，它主要是执行某个init的过程，然后又执行了某个register的过程。

第二个 doBind0 方法，主要是执行了端口的绑定，然后创建了eventLoop不断的执行JDK中的Selector.select()方法，从注册到selector中的channel中选择符合条件的channel。另外创建了一个task，用来从选中的channel中读取数据，然后把读取到的数据给到childHandler进行处理。

下面让我们来深入到这两个方法的执行过程中去，看看到底发生了什么。

5.1 执行initAndRegister方法

initAndRegister方法的执行过程如下图所示：

图说Netty服务端启动过程

initAndRegister方法做的事有两件：init和register。在这之前首先通过channelFactory创建了一个channel。该方法是在初始化EventLoopGroup的时候出现的，可以回头看一下，初始化的过程一共创建了三种对象：channel、unsafe、pipeline。

从该方法中慢慢的往下看，就可以看到，通过channelFactory创建了一个channel对象后，然后又拆分成了两个部分，分别对channel进行了初始化，和对channel进行了register。其中register方法，最终会调用到JDK中最原始的register方法，即把一个channel注册到一个selector中去。

  • init

初始化的过程主要是把用户先前创建的handler和childHandler添加到pipeline中去。

  • register

注册的过程主要是把该channel注册到selector中去，这里的channel就是用来接受客户端连接的。

5.2 执行doBind0方法

doBind0方法的执行过程如下图所示：

图说Netty服务端启动过程

doBind0做的事也很明确：bind、select以及runTask。

bind的过程最终是调用到JDK中原生的bind方法，其中在unsafe中执行bind的过程时，除了执行了具体的bind之外，还在NioEventLoop中启动了一个线程，用来不断的执行JDK中selector的select方法。然后读取选中的channel中的数据，最后把读取到的数据丢给childHandler去处理。

JDK的epoll空轮询bug

我们知道JDK中的Selector会出现epoll空轮询的bug，若Selector的轮询结果为空，也没有wakeup或新消息处理，则发生空轮询，此时CPU使用率将达到100%。

Netty是通过重建Selector的方式修复该bug的，具体的做法是：

  • 对Selector的select操作周期进行统计，每完成一次空的select操作进行一次计数，

  • 若在某个周期内连续发生n(SELECTORAUTOREBUILD_THRESHOLD)次空轮询，则触发了epoll死循环bug。

  • 重建Selector，判断是否是其他线程发起的重建请求，若不是则将原SocketChannel从旧的Selector上解除注册，重新注册到新的Selector上，并将原来的Selector关闭。

具体的代码是在NioEventLoop中的select方法中执行的，代码如下：

 
   
   
 
    
  
    
    
  
  1. private void select(boolean oldWakenUp) throws IOException {
    2. 
  
    
    
  
  2.     Selector selector = this.selector;
    3. 
  
    
    
  
  3.     try {
    4. 
  
    
    
  
  4.         int selectCnt = 0;
    5. 
  
    
    
  
  5.         long currentTimeNanos = System.nanoTime();
    6. 
  
    
    
  
  6.         long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
    7. 
  
    
    
  
  7.         for (;;) {
    8. 
  
    
    
  
  8.             long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
    9. 
  
    
    
  
  9.             if (timeoutMillis <= 0) {
    10. 
  
    
    
  
  10.                 if (selectCnt == 0) {
    11. 
  
    
    
  
  11.                     selector.selectNow();
    12. 
  
    
    
  
  12.                     selectCnt = 1;
    13. 
  
    
    
  
  13.                 }
    14. 
  
    
    
  
  14.                 break;
    15. 
  
    
    
  
  15.             }
    16. 
  
    
    
  
    17. 
  
    
    
  
  17.             if (hasTasks() && wakenUp.compareAndSet(false, true)) {
    18. 
  
    
    
  
  18.                 selector.selectNow();
    19. 
  
    
    
  
  19.                 selectCnt = 1;
    20. 
  
    
    
  
  20.                 break;
    21. 
  
    
    
  
  21.             }
    22. 
  
    
    
  
    23. 
  
    
    
  
  23.             int selectedKeys = selector.select(timeoutMillis);
    24. 
  
    
    
  
  24.             selectCnt ++;
    25. 
  
    
    
  
    26. 
  
    
    
  
  26.             if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
    27. 
  
    
    
  
  27.                 break;
    28. 
  
    
    
  
  28.             }
    29. 
  
    
    
  
  29.             if (Thread.interrupted()) {
    30. 
  
    
    
  
  30.                 selectCnt = 1;
    31. 
  
    
    
  
  31.                 break;
    32. 
  
    
    
  
  32.             }
    33. 
  
    
    
  
    34. 
  
    
    
  
  34.             long time = System.nanoTime();
    35. 
  
    
    
  
  35.             if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
    36. 
  
    
    
  
  36.                 // timeoutMillis elapsed without anything selected.
    37. 
  
    
    
  
  37.                 selectCnt = 1;
    38. 
  
    
    
  
  38.             // 当发生的select次数大于指定的阈值时，重建Selector    
    39. 
  
    
    
  
  39.             } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
    40. 
  
    
    
  
  40.                     selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
    41. 
  
    
    
  
  41.                 // 重建Selector，以解决JDK中的epoll的bug
    42. 
  
    
    
  
  42.                 rebuildSelector();
    43. 
  
    
    
  
  43.                 selector = this.selector;
    44. 
  
    
    
  
    45. 
  
    
    
  
  45.                 // Select again to populate selectedKeys.
    46. 
  
    
    
  
  46.                 selector.selectNow();
    47. 
  
    
    
  
  47.                 selectCnt = 1;
    48. 
  
    
    
  
  48.                 break;
    49. 
  
    
    
  
  49.             }
    50. 
  
    
    
  
    51. 
  
    
    
  
  51.             currentTimeNanos = time;
    52. 
  
    
    
  
  52.         }
    53. 
  
    
    
  
    54. 
  
    
    
  
  54.     } catch (CancelledKeyException e) {
    55. 
  
    
    
  
  55.         if (logger.isDebugEnabled()) {
    56. 
  
    
    
  
  56.             logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
    57. 
  
    
    
  
  57.                     selector, e);
    58. 
  
    
    
  
  58.         }
    59. 
  
    
    
  
  59.         // Harmless exception - log anyway
    60. 
  
    
    
  
  60.     }
    61. 
  
    
    
  
  61. }
    62.

完整的启动过程

通过上面的分析，我们最后来总结一下，Netty服务端在启动的时候做了以下的事情：

  • 1.创建了EventLoopGroup、NioEventLoop的实例，并且创建了一个selector

  • 2.创建了一个channelHandler用来在未来实例化Channel

    • 创建Channel的过程中会一并创建pipeline和unsafe

  • 3.设置了ServerBootstrap的handler和childHandler属性，用以在接收到数据后进行业务逻辑的处理

  • 4.通过channelFactory创建了channel实例，并对其进行了初始化和注册到selector上

  • 5.通过Unsafe调用JDK的bind方法将服务绑定到了端口上，并通过EventLoop创建了一个线程来循环执行以下任务

    • 5.1.执行selector的select方法，并通过计数的方式，满足一定条件的情况下对selector进行重建，以解决JDK的epoll空轮询的bug

    • 5.2.对选中的channel执行读操作，并将读取到的数据丢给childHandler进行处理

一个完整的Netty服务端启动过程如下图所示：



以上是关于[Netty源码] 服务端启动过程的主要内容，如果未能解决你的问题，请参考以下文章

图说Netty服务端启动过程

Netty源码分析之服务端启动

netty源码分析之服务端启动全

死磕Netty-----服务端启动过程分析

说说Netty服务端启动流程

Netty 线程服务端启动流程