netty服务端启动过程分析

Love The Way You Lie 2023-06-13 04:42 17阅读 0赞

# netty 服务端启动 #

1.  创建ServerBootstrap,该类是Netty服务端的启动类
2.  绑定Reactor线程池,EventLoopGroup,实际上就是EventLoop的数组.除了处理IO操作外,用户提交的task和定时任务也是由EventLoop执行.避免了多线程竞争的情况
3.  设置并绑定服务端Channel,对于Nio服务端,使用的是NioServerSocketChannel
4.  链路建立是创建并初始化ChannelPipeline.其本质是一个处理网络事件的职责链
5.  初始化ChannelPipline完成之后,添加并设置ChannelHandler,
6.  绑定并启动监听端口,在绑定之前系统会先进行一些初始化,完成之后会启动监听端口,并将ServerSocketChannel注册到Selector上监听客户端连接
7.  Selector轮询.由Reactor线程NioEventLoop负责调度和执行Selector轮询,选择准备就绪的Channel集合.
8.  当轮询到准备就绪的Channel之后，由Reactor线程执行ChannelPipeline的相应方法，最终调度执行ChannelHandler
9.  执行Netty系统ChannelHandler和用户自定义的ChannelHandler。ChannelPipeline根据网络事件的类型，调度并执行ChannelHandler。

## 源码分析 ##

服务端从bind 函数启动。bind 函数最后访问到了`AbstractBootstrap`的 dobind 函数，如下

private ChannelFuture doBind(final SocketAddress localAddress) {
      final ChannelFuture regFuture = initAndRegister();
      //省略其他函数
    }

### 初始化 ###

final ChannelFuture initAndRegister() {
            Channel channel = null;
            try {
                channel = channelFactory.newChannel();
                init(channel);
            //省略其他函数
        }

1.  使用 channelFactory 的 newChannel 构造一个 channel，对于 Nio应用来说，这里是`NioServerSocketChannel`，其使用了反射来获取实例。
2.  调用 `init` 函数

init 函数是一个抽象函数，被定义在了`ServerBootstrap`中

void init(Channel channel) throws Exception { 
        final Map<ChannelOption<?>, Object> options = options0();
      设置 Socket 参数和 NioServerSocketChannel 的附加属性
        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);///将 AbstractBootstrap 的 Handler 添加到 ChannelPipeline 中
                }
    
                ch.eventLoop().execute(new Runnable() { 
                    @Override
                    public void run() { 
                      //添加用于注册的 handler
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

前面一大堆其实都是设置连接参数，对于 Server 来说关键点在于添加了`ServerBootstrapAcceptor`这是一个内部类，并且其是server 的唯一一个handler。暂且略过不提。

### 注册 ###

init 后回到`initAndRegister`函数

final ChannelFuture initAndRegister() {
      //省略其他代码
      ChannelFuture regFuture = config().group().register(channel);
      if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
          channel.close();
        } else {
          channel.unsafe().closeForcibly();
        }
      }
      return regFuture;
    }

从当前的 执行线程中获取一个并将 channel注册到这个线程中。

对于 NioServerChannel其调用到了SingleThreadEventLoop的 register 函数

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

对于 NioServerChannel 的 unsafe 函数，其在AbstractNioMessageChannel中实现

而 register 函数，则在AbstractUnsafe中定义，并被声明为 final

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;
    
      if (eventLoop.inEventLoop()) { 
        register0(promise);
      } else { 
        try { 
          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);
        }
      }
            }

前面的函数是参数校验，

关键代码为

AbstractChannel.this.eventLoop = eventLoop;

这行代码将 evetloop（即指定的线程） 绑定到了 channel 中。

此时调用线程仍然是用户线程，因此会进入

eventLoop.execute(new Runnable() { 
        @Override
        public void run() { 
            register0(promise);
        }
    });

从此刻开始，**netty 线程和用户线程就开始并行**了。

随后由 eventLoop 来执行`register0`函数

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);
        }
    }

#### AbstractNioChannel.doRegister ####

执行 doRegister 函数，其被定义在`AbstractNioChannel`中

protected void doRegister() throws Exception { 
      boolean selected = false;
      for (;;) { 
        try { 
          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;
          }
        }
      }
    }

其 javaChannel 由`NioServerSocketChannel`的构造器设置。

这里注册的是 0，表示不监听任何网络操作。

这里注册为 0 的原因有两点

1.  该函数处于`AbstractNioChannel`，而 clientChannel 和 serverChannel 都是其子类，因此其不能设置特殊的操作
2.  通过 SelectionKey 的 interestOps 可以修改其监听的网络操作

#### other ####

pipeline 实际上是一个AbstractChannelHandlerContext的双向链表，并且其默认的 head 和 tail 分别为HeadContext和TailContext。

headContext的 register 的实现为invokeHandlerAddedIfNeeded，即触发第一次注册事件，然而此时其实是不触发的，因为刚刚在之前已经触发过了，随后将在链表中传递注册事件

tailContext 是一个空操作。

##### 初始化 server #####

除此之外，在初始化 server 时，还添加了一个 `ChannelInitializer`（在 init 函数中）

查看其实现

@Override
    @SuppressWarnings("unchecked")
    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception { 
        // Normally this method will never be called as handlerAdded(...) should call initChannel(...) and remove
        // the handler.
        if (initChannel(ctx)) { 
            // we called initChannel(...) so we need to call now pipeline.fireChannelRegistered() to ensure we not
            // miss an event.
            ctx.pipeline().fireChannelRegistered();
    
            // We are done with init the Channel, removing all the state for the Channel now.
            removeState(ctx);
        } else { 
            // Called initChannel(...) before which is the expected behavior, so just forward the event.
            ctx.fireChannelRegistered();
        }
    }
    private boolean initChannel(ChannelHandlerContext ctx) throws Exception { 
      if (initMap.add(ctx)) {  // Guard against re-entrance.
        try { 
          initChannel((C) ctx.channel());
        } catch (Throwable cause) { 
          // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
          // We do so to prevent multiple calls to initChannel(...).
          exceptionCaught(ctx, cause);
        } finally { 
          ChannelPipeline pipeline = ctx.pipeline();
          if (pipeline.context(this) != null) { 
            pipeline.remove(this);
          }
        }
        return true;
      }
      return false;
    }

可以看到其在触发 register 事件触发时的默认操作为执行`initChannel`操作，随后结束移除 init 的上下文这个操作是自定义的，对于 server 来说这个操作为

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));
        }
    });

在这里又提交了一个 task 到 channel 绑定的 eventLoop 中，实际上当前正在进行的 register 也是一个 task，因此这里提交了下一个运行的任务。

需要注意的是**这个提交的任务并不一定在下次执行**，因为当前是 netty 线程执行，而用户线程可能是并行的。

随后调用 isActive，对于 Server 来说，其实现于NioServerSocketChannel

public boolean isActive() { 
      // As java.nio.ServerSocketChannel.isBound() will continue to return true even after the channel was closed
      // we will also need to check if it is open.
      return isOpen() && javaChannel().socket().isBound();
    }

随后触发 pipeline 的 active 操作，对于 head 来说在触发了链表的所有 active 操作后，若配置为 autoRead，则进行 channel的 read 操作

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

随后若当前 channel 注册过，则触发beginRead 函数

此时将调用到AbstractNioChannel 的doBeginRead 操作，其将设置对 channel 添加一个 read 的interestOps。

@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.interestOps(interestOps | readInterestOp);
        }
    }

然而对于NioServerSocketChannel来说，readInterestOps 为OP\_ACCEPT

public NioServerSocketChannel(ServerSocketChannel channel) { 
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

### 绑定 ###

当触发并行时用户线程回到 bind 函数中

private ChannelFuture doBind(final SocketAddress localAddress) {
      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();
        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;
      }
    }

1.  确定 initAndRegister 没有发生异常
2.  判断 register 是否完成

#### 注册完成时 ####

若当前 register 已经完成了，则直接进行 bind0

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，并且对 bind 事件添加失败关闭事件，要注意的是这里是给 channel 绑定的线程提交了一个任务，而非直接执行，bind 函数具体的操作暂且不提。

ChannelFutureListener CLOSE_ON_FAILURE = new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) {
        if (!future.isSuccess()) {
          future.channel().close();
        }
      }
    };

#### 注册未完成时 ####

给注册事件添加一个完成时事件

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);
        }
      }
    });

最后仍然是执行 doBind0.

#### doBind0 ####

bind 函数调用到AbstractChannel

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

DefaultChannelPipeline 的 bind

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

TailContext 的 bind由AbstractChannelHandlerContext 实现

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(MASK_BIND);
        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;
    }

当前情况这里运行的线程为 eventLoop

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) { 
        if (invokeHandler()) { //期望值为已完成 handler add 事件
            try { 
                ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
            } catch (Throwable t) { 
                notifyOutboundHandlerException(t, promise);
            }
        } else { //递归回bind函数
            bind(localAddress, promise);
        }
    }

实际上由于 eventloop 是单线程的，因此执行到这里时，必定已经注册完成了，也就是

调用回 HeadContext

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

由 AbstractUnsafe 实现

@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 { 
            doBind(localAddress);
        } catch (Throwable t) { 
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
    
        if (!wasActive && isActive()) { 
            invokeLater(new Runnable() { 
                @Override
                public void run() { 
                    pipeline.fireChannelActive();
                }
            });
        }
    
        safeSetSuccess(promise);
    }

doBind函数由 NioServerChannel 实现，这里是由 eventLoop 执行的

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

随后若开始doBind前未 active，并且 bind 后active，则提交一个activefire任务到 eventloop 中。

至此服务端启动完成