30.Netty源码服务端启动主要流程

news2024/11/24 13:27:17

highlight: arduino-light

服务端启动主要流程

•创建 selector

•创建 server socket channel

•初始化 server socket channel

•给 server socket channel 从 boss group 中选择一个 NioEventLoop

•将 server socket channel 注册到选择的 NioEventLoop 的 selector

•绑定地址启动

•注册接受连接事件(OP_ACCEPT)到 selector上

从Echo服务器示例入手

在《引导器作用:客户端和服务端启动都要做些什么?》的课程中,我们介绍了如何使用引导器搭建服务端的基本框架。在这里我们实现了一个最简单的 Echo 服务器,用于调试 Netty 服务端启动的源码。

java public class EchoServer {    public void startEchoServer(int port) throws Exception {   //默认会创建 cpu核心数*2 个 NioEventLoop   //创建NioEventLoop 会调用openSelector 创建一个Selector   //将创建的Selector赋值给NioEventLoop的selector属性        EventLoopGroup bossGroup = new NioEventLoopGroup();        EventLoopGroup workerGroup = new NioEventLoopGroup();        try {            ServerBootstrap b = new ServerBootstrap();            b.group(bossGroup, workerGroup)                   .channel(NioServerSocketChannel.class)               .handler(new LoggingHandler(LogLevel.INFO))                   .childHandler(new ChannelInitializer<SocketChannel>() {                        @Override                        public void initChannel(SocketChannel ch) {                            ch.pipeline().addLast                               (new FixedLengthFrameDecoder(10));                            ch.pipeline().addLast(new ResponseSampleEncoder());                            ch.pipeline().addLast(new RequestSampleHandler());                       }                   });            //同步阻塞            ChannelFuture f = b.bind(port).sync();            f.channel().closeFuture().sync();       } finally {            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();       }   }    public static void main(String[] args) throws Exception {        new EchoServer().startEchoServer(8088);   } } ​ public class ResponseSampleEncoder extends MessageToByteEncoder<ResponseSample> {    @Override    protected void encode(ChannelHandlerContext ctx, ResponseSample msg, ByteBuf out) { ​        if (msg != null) {            out.writeBytes(msg.getCode().getBytes());            out.writeBytes(msg.getData().getBytes());            out.writeLong(msg.getTimestamp());       }   } } ​ public class RequestSampleHandler extends ChannelInboundHandlerAdapter {    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) {        String data = ((ByteBuf) msg).toString(CharsetUtil.UTF_8);        ResponseSample response = new ResponseSample("OK", data,                                                     System.currentTimeMillis());        ctx.channel().writeAndFlush(response);   } } ​

```java 我们以引导器 ServerBootstrap 为切入点,开始深入分析 Netty 服务端的启动流程。在服务端启动之前,需要配置 ServerBootstrap 的相关参数,这一步大致可以分为以下几个步骤:

配置 EventLoopGroup 线程组;

配置 Channel 的类型;

设置初始化Handler;

设置网络监听的端口;

设置处理Handler;

配置 Channel 参数。 ```

配置 ServerBootstrap 参数的过程非常简单,把参数值保存在 ServerBootstrap 定义的成员变量里就可以了。我们可以看下 ServerBootstrap 的成员变量定义,基本与 ServerBootstrap 暴露出来的配置方法是一一对应的。如下所示,我以注释的形式说明每个成员变量对应的调用方法。

java volatile EventLoopGroup group; // group() volatile EventLoopGroup childGroup; // group() volatile ChannelFactory<? extends C> channelFactory; // channel() volatile SocketAddress localAddress; // localAddress Map<ChannelOption<?>, Object> childOptions = new ConcurrentHashMap<ChannelOption<?>, Object>(); // childOption() volatile ChannelHandler childHandler; // childHandler() ServerBootstrapConfig config = new ServerBootstrapConfig(this);

关于 ServerBootstrap 如何为每个成员变量保存参数的过程,我们就不一一展开了,你可以理解为这部分工作只是一个前置准备,课后你可以自己跟进下每个方法的源码。

今天我们核心聚焦在 b.bind().sync() 这行代码,bind() 才是真正进行服务器端口绑定和启动的入口,sync() 表示阻塞等待服务器启动完成。接下来我们对 bind() 方法进行展开分析。

在开始源码分析之前,我们带着以下几个问题边看边思考:

Netty 自己实现的 Channel 与 JDK 底层的 Channel 是如何产生联系的?

ChannelInitializer 这个特殊的 Handler 处理器的作用是什么?

Pipeline 初始化的过程是什么样的?

服务端启动全过程

入口:bind方法

首先我们来看下 ServerBootstrap 中 bind() 方法的源码实现

java //bind方法入口 public ChannelFuture bind() { validate(); //已通过 SocketAddress localAddress = this.localAddress; if (localAddress == null) { throw new IllegalStateException("localAddress not set"); } //doBind return doBind(localAddress); }

```java private ChannelFuture doBind(final SocketAddress localAddress) { //初始化并注册 Channel,同时返回一个 ChannelFuture 实例 regFuture, //所以我们可以猜测出
//initAndRegister() 是一个异步的过程。 final ChannelFuture regFuture = initAndRegister();

final Channel channel = regFuture.channel();
//判断 initAndRegister() 的过程是否发生异常,如果发生异常则直接返回。
if (regFuture.cause() != null) {
    return regFuture;
}

//regFuture.isDone() 即initAndRegister() 是否执行完毕 //如果执行完毕则调用doBind0() 进行Socket 绑定 //如果 initAndRegister() 还没有执行结束,regFuture 会添加ChannelFutureListener 回调监听 //当 initAndRegister() 执行结束后会调用 operationComplete(),同样通过 doBind0() 进行端口绑定。 if (regFuture.isDone()) { ChannelPromise promise = channel.newPromise(); //doBind0() 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(); //regFuture.isDone() 如果没有执行完毕,这里的监听器会调用doBind0 doBind0(regFuture, channel, localAddress, promise); } } }); return promise; } } ```

doBind0() 整个实现结构非常清晰,其中 initAndRegister() 负责 NioSocketServerChannel初始化和注册,doBind0() 用于端口绑定。这两个过程最为重要,下面我们分别进行详细的介绍。

InitAndRegister方法

initAndRegister() 方法顾名思义,主要负责创建NioServerSocketChannel初始化NioServerSocketChannel和注册NioServerSocketChannel的相关工作,我们具体看下它的源码实现

java final ChannelFuture initAndRegister() { Channel channel = null; try { // 通过指定的NioSocketServerChannel.class 反射创建 // 1.创建 Channel channel = channelFactory.newChannel(); // 2.初始化 Channel init(channel); } catch (Throwable t) { if (channel != null) { channel.unsafe().closeForcibly(); return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t); } //3.注册Channel //config().group()是什么? //是bossGroup ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }

initAndRegister() 可以分为三步:创建 Channel、初始化 Channel 和注册 Channel,接下来我们一步步进行拆解分析。

创建服务器端Channel

1.创建NioServerSocketChannel

首先看下创建 Channel 的过程,直接跟进 channelFactory.newChannel() 的源码。

java public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> { private final Constructor<? extends T> constructor; public ReflectiveChannelFactory(Class<? extends T> clazz) { ObjectUtil.checkNotNull(clazz, "clazz"); try { //这里通过泛型反射+工厂 获取无参构造方法 //传进来的clazz是NioServerSocketChannel.class this.constructor = clazz.getConstructor(); } catch (NoSuchMethodException e) { throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) + " does not have a public non-arg constructor", e); } } @Override public T newChannel() { try { // 反射创建对象 return constructor.newInstance(); } catch (Throwable t) { throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t); } } // 省略其他代码 }

在前面 EchoServer的示例中,我们通过 channel(NioServerSocketChannel.class) 配置 Channel 的类型,工厂类 ReflectiveChannelFactory 是在该过程中被创建的。

从 constructor.newInstance() 我们可以看出,ReflectiveChannelFactory 通过反射创建出 NioServerSocketChannel 对象,所以我们重点需要关注 NioServerSocketChannel 的构造函数。

```java //private static final SelectorProvider //DEFAULTSELECTORPROVIDER = SelectorProvider.provider(); public NioServerSocketChannel() { //DEFAULTSELECTORPROVIDER: 根据不同的系统返回不同的SelectorProvider // 很熟悉啊,newSocket(DEFAULTSELECTORPROVIDER)是创建 JDK 底层的 ServerSocketChannel this(newSocket(DEFAULTSELECTORPROVIDER)); }

//根据不同的 SelectorProvider 创建不同的JDK 底层的 ServerSocketChannel private static ServerSocketChannel newSocket(SelectorProvider provider) { try { // 创建 JDK 底层的 ServerSocketChannel 实现类是ServerSocketChannelImpl
return provider.openServerSocketChannel(); } catch (IOException e) { throw new ChannelException( "Failed to open a server socket.", e); } }

//将JDK 底层的 ServerSocketChannel 包装为NioServerSocketChannel 并注册OPACCEPT事件 public NioServerSocketChannel(ServerSocketChannel channel) { // 调用父类AbstractChannel方法 // 注意这里是SelectionKey.OPACCEPT=16 // 并不是注册ACCEPT事件 // 只是设置 this.readInterestOp = readInterestOp; 即设置为16 super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config = new NioServerSocketChannelConfig(this, javaChannel().socket()); } ```

SelectorProvider 是 JDK NIO 中的抽象类实现,通过 openServerSocketChannel() 方法可以用于创建服务端的 ServerSocketChannel。而且 SelectorProvider 会根据操作系统类型和版本的不同,返回不同的实现类,具体可以参考 DefaultSelectorProvider 的源码实现:

java public static SelectorProvider provider() { synchronized (lock) { if (provider != null) return provider; return AccessController.doPrivileged( new PrivilegedAction<SelectorProvider>() { public SelectorProvider run() { //1.读取配置根据配置的class获取provider 獲取不到到第二步 if (loadProviderFromProperty()) return provider; //2.通过spi获取provider 获取不到到第三步 if (loadProviderAsService()) return provider; //3.DefaultSelectorProvider#create创建provider //根据不同的系统创建不同的Selector 或者是说jdk不同 //Linux下JDK的下载和安装与Windows下并没有太大的不同 //只是对一些环境的设置稍有不同。 //在windows环境下的是WindowsSelectorProvider provider = sun.nio.ch.DefaultSelectorProvider.create(); return provider; } }); } }

看一下sun.nio.ch.DefaultSelectorProvider.create();

java public static SelectorProvider create() { String osname = AccessController .doPrivileged(new GetPropertyAction("os.name")); if (osname.equals("SunOS")) return createProvider("sun.nio.ch.DevPollSelectorProvider"); if (osname.equals("Linux")) return createProvider("sun.nio.ch.EPollSelectorProvider"); //默认返回的是PollSelectorProvider return new sun.nio.ch.PollSelectorProvider(); }

在这里我们只讨论 Linux 操作系统的场景,在 Linux 内核 2.6版本及以上都会默认采用 EPollSelectorProvider。

如果是旧版本则使用 PollSelectorProvider。对于目前的主流 Linux 平台而言,都是采用 Epoll 机制实现的。

创建完JDK的ServerSocketChannel,我们回到 NioServerSocketChannel 的构造函数

java //将JDK 底层的 ServerSocketChannel 包装为NioServerSocketChannel 并注册OP_ACCEPT事件 public NioServerSocketChannel(ServerSocketChannel channel) { // 调用父类AbstractChannel方法 // 注意这里是SelectionKey.OP_ACCEPT=16 // 并不是注册ACCEPT事件 // 只是设置 this.readInterestOp = readInterestOp; 即设置为16 super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config = new NioServerSocketChannelConfig(this, javaChannel().socket()); }

接着它会通过依次调用到父类的构造进行初始化工作

java super(null, channel, SelectionKey.OP_ACCEPT);

最终我们可以定位到 AbstractNioChannel 和 AbstractChannel 的构造函数:

java protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); // 省略其他代码 //设置为16 this.readInterestOp = readInterestOp; try { //设置非阻塞 ch.configureBlocking(false); } catch (IOException e) { // 省略其他代码 } } protected AbstractChannel(Channel parent) { this.parent = parent; // Channel 全局唯一 id id = newId(); // unsafe 操作底层读写 unsafe = newUnsafe(); // pipeline 负责业务处理器编排 // 会初始化TailContext和HeadContext pipeline = newChannelPipeline(); }

2.设置pipeline

首先调用 AbstractChannel 的构造函数创建了三个重要的成员变量,分别为 id、unsafe、pipeline。

id 表示全局唯一的 Channel,

unsafe 用于操作底层数据的读写操作,

pipeline 负责业务处理器的编排。

3.设置非阻塞模式

初始化状态,pipeline 的内部结构只包含头尾两个节点,如下图所示。三个核心成员变量创建好之后,会回到 AbstractNioChannel 的构造函数,通过 ch.configureBlocking(false) 设置 Channel 是非阻塞模式。

netty17图.png

创建服务端 Channel 的过程我们已经讲完了,简单总结下其中几个重要的步骤:

java ReflectiveChannelFactory 通过反射创建 NioServerSocketChannel 实例; 创建JDK底层的ServerSocketChannel,调用NioServerSocketChannel的构造函数,将底层的ServerSocketChannel包装为 NioServerSocketChannel。 在构造函数中,执行以下逻辑 1.为 Channel 创建 id、unsafe、pipeline 三个重要的成员变量; 2.设置 Channel 为非阻塞模式。 返回 NioServerSocketChannel。

初始化服务器端Channel

回到 ServerBootstrap 的 initAndRegister() 方法,继续跟进用于初始化服务端 Channel 的 init() 方法源码:

注意Channel是NioServerSocketChannel

java void init(Channel channel) { // 设置 Socket 参数,用户自定义的参数都放在了名为options的map中,此处是遍历map设置属性 // 底层也是做判断设置的 好low啊 setChannelOptions (channel, options0().entrySet().toArray(newOptionArray(0)), logger); // 保存用户自定义属性 setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0))); //获取pipeline ChannelPipeline p = channel.pipeline(); //下面的四个参数是为了childOption 即 NioSocketChannel // 也是获取 ServerBootstrapAcceptor 的构造参数 final EventLoopGroup currentChildGroup = childGroup; final ChannelHandler currentChildHandler = childHandler; final Entry<ChannelOption<?>, Object>[] currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0)); final Entry<AttributeKey<?>, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0)); //class ChannelInitializer extends ChannelInboundHandlerAdapter //ChannelInitializer是一个ChannelInboundHandlerAdapter //添加特殊的Handler处理器ChannelInitializer 它是一个一次性初始化的hanlder //负责添加一个ServerBootstrapAcceptor handler,添加完后,自己就移除了: //ServerBootstrapAcceptor handler: 负责接收客户端连接创建连接后,对连接的初始化工作。 p.addLast(new ChannelInitializer<Channel>() { @Override public void initChannel(final Channel ch) { final ChannelPipeline pipeline = ch.pipeline(); //添加配置的handler即.handler(new LoggingHandler(LogLevel.INFO)) ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } //这里还没有和具体的eventLoop绑定 //需要等待NioServerSocketChannel注册完成 //即reigister方法调用完成才拥有eventLoop //execute方法是往taskQueue.offer(task); ch.eventLoop().execute(new Runnable() { @Override public void run() { //构造ServerBootstrapAcceptor //负责接收客户端连接对连接的初始化工作。 //主要是接收到连接以后 为连接设置 option hanlder 设置属性 //currentChildHandler对应的是 wokrGroup的.childHandler()方法 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }

init() 方法的源码比较长,我们依然拆解成两个部分来看:

1.设置OPTION参数

第一步,设置 Socket 参数以及用户自定义属性。在创建服务端 Channel 时,Channel 的配置参数保存在 NioServerSocketChannelConfig 中,在初始化 Channel 的过程中,Netty 会将这些参数设置到 JDK 底层的 Socket 上,并把用户自定义的属性绑定在 Channel 上。

java public NioServerSocketChannel(ServerSocketChannel channel) { super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config = new NioServerSocketChannelConfig(this, javaChannel().socket()); }

2.添加匿名ChannelInitializer

```java @Override void init(Channel channel) { setChannelOptions (channel, options0().entrySet().toArray(newOptionArray(0)), logger); setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0)));

ChannelPipeline p = channel.pipeline();

    final EventLoopGroup currentChildGroup = childGroup;
    final ChannelHandler currentChildHandler = childHandler;
    final Entry<ChannelOption<?>, Object>[] currentChildOptions =
                        childOptions.entrySet().toArray(newOptionArray(0));
    final Entry<AttributeKey<?>, Object>[] currentChildAttrs = 
                        childAttrs.entrySet().toArray(newAttrArray(0));
    //ChannelInitializer一次性、初始化handler:
    //负责添加一个ServerBootstrapAcceptor handler,添加完后,自己就移除了:
    //ServerBootstrapAcceptor handler: 负责接收客户端连接创建连接后,对连接的初始化工作。
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) {
           //获取pipeline
            final ChannelPipeline pipeline = ch.pipeline();
             //获取handler方法中指定的handler
            ChannelHandler handler = config.handler();
            //如果handler不为空 加入 pipeline
            if (handler != null) {
                pipeline.addLast(handler);
            }
            //添加1个任务
            //任务添加了ServerBootstrapAcceptor
            //在ServerBootstrapAcceptor中会将childHandler添加到pipline
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            ch, currentChildGroup,
                            currentChildHandler, 
                            currentChildOptions,
                            currentChildAttrs));
                }
            });
        }
    });
}

```

第二步:添加匿名ChannelInitializer。

1.在ServerBootstrap#init为 Pipeline 添加了一个 匿名ChannelInitializer,ChannelInitializer 是实现了 ChannelHandler 接口的匿名类,其中 ChannelInitializer 实现的 initChannel() 方法做了2件事:

1.将handler方法中指定的处理器加入到pipeline

2.通过task 的方式又向 Pipeline添加了一个处理器ServerBootstrapAcceptor

从 ServerBootstrapAcceptor 的命名可以看出,这是一个连接接入器,专门用于接收新的连接,然后把事件分发给 EventLoop 执行,在这里我们先不做展开。

此时服务端的 pipeline 内部结构发生了变化,如下图所示。

图片1.png

思考一个问题,为什么需要 ChannelInitializer 处理器呢?

ServerBootstrapAcceptor 的注册过程为什么又需要封装成异步 task 呢?

因为我们在初始化时,还没有将 Channel 注册到 Selector 对象上,所以还无法注册 Accept 事件到 Selector 上(只是eventLoop和Selector 做了绑定,selecor在创建eventLoop的时候调用openSelector方法创建的), 所以事先在pipeline添加了 ChannelInitializer 处理器,等待 Channel 注册完成后,再向 Pipeline 中添加 ServerBootstrapAcceptor 处理器。

服务端 Channel 初始化的过程已经结束了。

整体流程比较简单,主要是设置 Socket 参数以及用户自定义属性,并向 Pipeline 中添加了1个特殊的处理器。

接下来我们继续分析,如何将初始化好的 Channel 注册到 Selector 对象上?

注册服务端 Channel

回到 initAndRegister() 的主流程,创建完服务端 Channel 之后,继续一层层跟进 register() 方法的源码:

这里的逻辑是由bossGroup中的eventLoop处理。

```java // MultithreadEventLoopGroup#register public ChannelFuture register(Channel channel) { // 选择一个eventLoop注册 //next(): //return executors[Math.abs(idx.getAndIncrement() % executors.length)]; //其实就是选择一个eventLoop将channel注册上去 //注意eventLoop已经和selector绑定了 return next().register(channel); }

//SingleThreadEventLoop#register public ChannelFuture register(Channel channel) { //注意这里的channel被封装进入了一个 DefaultChannelPromise //this是MultithreadEventLoop return register(new DefaultChannelPromise(channel, this)); } //SingleThreadEventLoop#register public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise"); this是SingleThreadEventLoop promise.channel().unsafe().register(this, promise); return promise; }

// AbstractChannel#register public final void register(EventLoop eventLoop, final ChannelPromise promise) { // 省略其他代码 AbstractChannel.this.eventLoop = eventLoop; // 判断当前eventLoop是否是本线程调用 // boosGroup线程组线程内部调用 if (eventLoop.inEventLoop()) { register0(promise); } else { // 外部线程调用 try { eventLoop.execute(new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { // 省略其他代码 } } } ```

Netty 会在线程池 BossEventLoopGroup 中选择一个 EventLoop 与当前 Channel 进行绑定,之后 Channel 生命周期内的所有 I/O 事件都由这个 EventLoop 负责处理,如 accept、connect、read、write 等 I/O 事件。

可以看出,不管是 EventLoop 线程本身调用,还是外部线程用,最终都会通过 register0() 方法进行注册:

java private void register0(ChannelPromise promise) { try { if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; // 调用 JDK 底层的 register() 进行注册 doRegister(); neverRegistered = false; registered = true; //触发匿名处理器的handlerAdded事件 //底层调用了callHandlerAdded0 //并调用了匿名处理器地initChannel方法 // 1.向pipeline添加handler中指定的处理器 // 2.向pipeline添加ServerBootStrapAcceptor //注意这2个是针对服务器端的channel添加的处理器。 //然后接收到新连接以后会使用这2个处理器处理客户端channel //ServerBootStrapAcceptor会把childHandler方法中的处理器放到客户端channel的pipeline pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); // 触发 channelRegistered 事件 pipeline.fireChannelRegistered(); // 此时 Channel 还未注册绑定地址,所以处于非活跃状态 if (isActive()) { if (firstRegistration) { // Channel 当前状态为活跃时,触发 channelActive 事件 pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { // 省略其他代码 } }

register0() 主要做了四件事:

```java 1.调用 JDK 底层进行 Channel 注册、

2.触发 handlerAdded 事件、

3.触发 channelRegistered 事件、

4.Channel 当前状态为活跃时,触发 channelActive 事件。 ```

1.注册Channel到选择器和注册事件0

为什么注册0?因为还没初始化完成

我们对它们逐一进行分析。

首先看下 JDK 底层注册 Channel 的过程,对应 doRegister() 方法的实现逻辑。

java protected void doRegister() throws Exception { boolean selected = false; for (;;) { try { // 调用 JDK 底层的 register() 进行注册 // eventLoop().unwrappedSelector()指的是未包装的selector // 包装的selector指的是 selectKey // 注意这里注册的事件是 0 是 0 是 0 // 注意这里注册的事件是 0 是 0 是 0 // 注意这里注册的事件是 0 是 0 是 0 // this = NioServerSocketChannel selectionKey = javaChannel().register (eventLoop().unwrappedSelector(), 0, this); return; } catch (CancelledKeyException e) { // 省略其他代码 } } } public final SelectionKey register(Selector sel, int ops, Object att)throws ClosedChannelException{ synchronized (regLock) { // 这里是查询1个选择器并返回 SelectionKey k = findKey(sel); if (k != null) { //注册事件 0 k.interestOps(ops); //设置附件 //att = NioServerSocketChannel k.attach(att); } if (k == null) { synchronized (keyLock) { if (!isOpen()) throw new ClosedChannelException(); k = ((AbstractSelector)sel).register(this, ops, att); addKey(k); } } return k; } }

javaChannel().register() 负责调用 JDK 底层,将 Channel 注册到 Selector 上,register() 的第三个入参传入的是 Netty 自己实现的 NioServerSocketChannel 对象,调用 register() 方法会将NioServerSocketChannel 绑定在 JDK 底层 Channel 的 attachment 上。

这样在每次 Selector对象进行事件循环时,Netty都可以从返回的JDK底层Channel中获得自己的Channel对象。

2.第一次注册触发handlerAdded事件

完成 Channel 向 Selector 注册后,接下来就会触发 Pipeline 一系列的事件传播。

在事件传播之前,用户自定义的业务处理器是如何被添加到 Pipeline 中的呢?

答案:就在pipeline.invokeHandlerAddedIfNeeded() 当中。

java final void invokeHandlerAddedIfNeeded() { assert channel.eventLoop().inEventLoop(); //只有第一次注册才会调用该方法 否则不会调用该方法 if (firstRegistration) { firstRegistration = false; System.out.println("invokeHandlerAddedIfNeeded:添加处理器"); callHandlerAddedForAllHandlers(); } }

我们重点看下 handlerAdded 事件的处理过程。invokeHandlerAddedIfNeeded() 方法的调用层次比较深,推荐你结合上述 Echo 服务端示例,使用 IDE Debug 的方式跟踪调用栈,如下图所示。

```java private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) { try { ctx.setAddComplete(); //ctx.handler()只有1个就是我们初始化的时候添加的匿名ChannelInitializer //调用对应ChannelInitializer的handlerAdded ctx.handler().handlerAdded(ctx); } catch (Throwable t) { boolean removed = false; try { remove0(ctx); try { ctx.handler().handlerRemoved(ctx); } finally { ctx.setRemoved(); } removed = true; } catch (Throwable t2) { if (logger.isWarnEnabled()) { logger.warn("Failed to remove a handler: " + ctx.name(), t2); } }

if (removed) {
            fireExceptionCaught(new ChannelPipelineException(
                    ctx.handler().getClass().getName() +
                    ".handlerAdded() has thrown an exception; removed.", t));
        } else {
            fireExceptionCaught(new ChannelPipelineException(
                    ctx.handler().getClass().getName() +
                    ".handlerAdded() has thrown an exception; also failed to remove.", t));
        }
    }
}

```

我们首先抓住 ChannelInitializer 中的核心源码,逐层进行分析。

java // ChannelInitializer public void handlerAdded(ChannelHandlerContext ctx) throws Exception { if (ctx.channel().isRegistered()) { //调用初始化方法 if (initChannel(ctx)) { removeState(ctx); } } } //ChannelInitializer private boolean initChannel(ChannelHandlerContext ctx) throws Exception { if (initMap.add(ctx)) { try { //调用匿名ChannelInitializer重写的initChannel()方法 //1.向 Pipeline 添加 handler方法中指定的handler //2.通过异步任务添加ServerBootstrapAcceptor处理器 initChannel((C) ctx.channel()); } catch (Throwable cause) { exceptionCaught(ctx, cause); } finally { ChannelPipeline pipeline = ctx.pipeline(); if (pipeline.context(this) != null) { // 将 ChannelInitializer 自身从 Pipeline 中移出 pipeline.remove(this); } } return true; } return false; }

可以看出 ChannelInitializer 首先会调用 initChannel() 抽象方法。

然后 Netty 会把 ChannelInitializer 自身从 Pipeline移除。

其中 initChannel() 抽象方法是在哪里实现的呢?

这就要跟踪到 ServerBootstrap 的 init() 方法,其中有这么一段代码:

java //此处的channel是服务器端的channel p.addLast(new ChannelInitializer<Channel>() { @Override public void initChannel(final Channel ch) { final ChannelPipeline pipeline = ch.pipeline //添加的是handler方法中指定的处理器。 //即服务器端配置的handler方法中指定的处理器。 //handler方法中指定的处理器在初始化时就会执行 //而childHandler方法中指定的处理器会在客户端成功connect后才执行,这是两者的区别。 ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { //handler方法中指定的处理器在初始化时就会执行, //而childHandler方法中指定的处理器会在客户端成功connect后才执行,这是两者的区别。 //ServerBootstrapAcceptor负责接收客户端连接。 //创建客户端连接后,对连接的初始化工作。 //因为添加 ServerBootstrapAcceptor是一个异步过程,需要EventLoop线程负责执行。 //而当前 EventLoop线程正在执行 register0() 的注册流程 //所以等到 register0() 执行完之后才能被添加到Pipeline当中。 //此时已经注册完成 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } });

3.调用匿名ChannelInitializer#initChannel

4.向服务器channel添加handler()中的handler

java ChannelHandler handler = config.handler(); if (handler != null) { System.out.println ("向服务器端添加serverBootstrap.handler()"+ "方法中指定的处理器ServerLoggingHandler"); pipeline.addLast(handler); }

5.触发LoggingHandler#handlerAdded

此时会触发处理器的handlerAdded事件,调用ChannelHandlerAdapter#handlerAdded的方法。

java final void callHandlerAdded() throws Exception { if (setAddComplete()) { //handler是当前被添加的处理器ServerLoggingHandler //ServerLoggingHandler继承了ChannelDuplexHandler //ChannelDuplexHandler extends ChannelInboundHandlerAdapter //所以调用的是ChannelInboundHandlerAdapter#handlerAdded handler().handlerAdded(this); } }

io.netty.channel.ChannelHandlerAdapter#handlerAdded

java @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //匿名ChannelInitializer继承自ChannelHandlerAdapter //在调用handlerAdded时会触发ChannelHandlerAdapter的handlerAdded方法 System.out.println("触发" + this.getClass().getSimpleName() + ".handlerAdded"); }

从代码的走势看应该添加 ServerBootstrapAcceptor 。

但是因为添加 ServerBootstrapAcceptor 是一个异步过程,需要 EventLoop 线程负责执行。

而当前 EventLoop 线程正在执行 register0() 的注册流程,所以等到 register0() 执行完之后才能被添加到 Pipeline 当中。也就是说添加ServerBootstrapAcceptor的runnable正在队列中等待被执行。

java 1.当前线程正在执行register0代码 2.当前线程代码中执行initChannel 3.当前线程然后添加handler方法中的处理器 4.当前线程中执行以下代码 ch.eventLoop().execute(new Runnable() { //添加SBAcceptor } 此时只是给线程添加了1个任务,然后放入了队列中 只有当当前线程把正在执行的register0代码执行完毕 才能继续把队列中的任务取出来执行

```java package io.netty.example.echo; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;

public class NettyThread { public static void main(String[] args) { final ExecutorService executorService = Executors.newFixedThreadPool(1); final Runnable runnable2= new Runnable() { @Override public void run() { System.out.println(2); } };

final Runnable runnable1= new Runnable() {
        @Override
        public void run() {
            System.out.println("我现在在执行register0开始");
            executorService.execute(runnable2);
            System.out.println("我现在在执行register0结束");
        }
    };
    executorService.execute(runnable1);
}

} /* 我现在在执行register0开始 我现在在执行register0结束 2 */ ```

完成 initChannel() 这一步之后,ServerBootstrapAcceptor 并没有被添加到 Pipeline 中,此时 Pipeline 的内部结构变化如下图所示。

Head --- LoggingHandler --- Tail

6.触发LoggingHandler的channelRegistered

我们回到 register0() 的主流程,接着向下分析。

channelRegistered 事件是由 fireChannelRegistered() 方法触发,沿着 Pipeline 的 Head 节点传播到 Tail 节点,并依次调用每个 ChannelHandler 的 channelRegistered() 方法。此时pipeline的示意图如下

Head --- LoggingHandler --- Tail

所以只会触发LoggingHandler的channelRegistered

java 触发事件:fireChannelRegistered ServerLoggingHandler.channelRegistered

然而此时 Channel 还未注册绑定地址,所以处于非活跃状态,所以并不会触发 channelActive 事件。

7. register0()的注册流程执行完毕

执行完整个 register0() 的注册流程之后。

EventLoop线程会将 ServerBootstrapAcceptor 添加到 Pipeline 当中。

java ch.eventLoop().execute(new Runnable() { @Override public void run() { System.out.println("服务器端异步添加ServerBootstrapAcceptor"); pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } });

此时会触发处理器的handlerAdded事件,调用ChannelHandlerAdapter#handlerAdded的方法。

java final void callHandlerAdded() throws Exception { if (setAddComplete()) { //handler是当前被添加的处理器ServerLoggingHandler //ServerLoggingHandler继承了ChannelDuplexHandler //ChannelDuplexHandler extends ChannelInboundHandlerAdapter //所以调用的是ChannelInboundHandlerAdapter#handlerAdded handler().handlerAdded(this); } }

io.netty.channel.ChannelHandlerAdapter#handlerAdded

java @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //匿名ChannelInitializer继承自ChannelHandlerAdapter //在调用handlerAdded时会触发ChannelHandlerAdapter的handlerAdded方法 System.out.println("触发" + this.getClass().getSimpleName() + ".handlerAdded"); }

此时 Pipeline 的内部结构又发生了变化,如下图所示。

image.png

8.向服务器channel添加SBAcceptor

在前面我们已经分析了 initChannel() 方法的实现逻辑,首先向 Pipeline 中添加 ServerSocketChannel 对应的 Handler,然后通过异步 task 的方式向 Pipeline 添加 ServerBootstrapAcceptor 处理器。

其中有一个点不要混淆,handler() 方法中的处理器和 ServerBootstrapAcceptor 处理器是添加到服务端的Pipeline 上。

而 childHandler() 方法是通过 ServerBootstrapAcceptor 处理器添加到客户端的 Pipeline 上。

9.端口绑定

整个服务端 Channel 注册的流程我们已经讲完,注册过程中 Pipeline 结构的变化值得你再反复梳理,从而加深理解。

目前服务端还是不能工作的,还差最后一步就是进行端口绑定,我们继续向下分析。

回到 ServerBootstrap 的 bind() 方法,我们继续跟进端口绑定 doBind0() 的源码。

java public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { assertEventLoop(); // 省略其他代码 boolean wasActive = isActive(); try { // 调用 JDK 底层进行端口绑定 doBind(localAddress); } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { invokeLater(new Runnable() { @Override public void run() { // 触发 channelActive 给ServerSocketChannel注册 // SelectionKey.OP_ACCEPT事件 // 所有事件的触发都是通过pipeline pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); }

bind() 方法主要做了两件事:

1.调用 JDK 底层进行端口绑定

2.绑定成功后并触发 channelActive 事件。下面我们逐一进行分析。

首先看下调用 JDK 底层进行端口绑定的 doBind() 方法:

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

Netty 会根据 JDK 版本的不同,分别调用 JDK 底层不同的 bind() 方法。我使用的是 JDK8,所以会调用 JDK 原生 Channel 的 bind() 方法。执行完 doBind() 之后,服务端 JDK 原生的 Channel 真正已经完成端口绑定了。

10.触发 channelActive 事件

完成端口绑定之后,Channel 处于活跃 Active 状态,然后会调用 pipeline.fireChannelActive() 方法触发 channelActive 事件。 即Channel 处于就绪状态,可以被读写。

我们可以一层层跟进 fireChannelActive() 方法,发现其中比较重要的部分:

java // DefaultChannelPipeline#channelActive public void channelActive(ChannelHandlerContext ctx) { ctx.fireChannelActive(); readIfIsAutoRead(); } // AbstractNioChannel#doBeginRead 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) { // 注册 OP_ACCEPT 事件到服务端 Channel 的事件集合 selectionKey.interestOps(interestOps | readInterestOp); } }

可以看出,在执行完 channelActive 事件传播之后,会调用 readIfIsAutoRead() 方法触发 Channel 的 read 事件,而它最终调用到 AbstractNioChannel 中的 doBeginRead() 方法,其中 readInterestOp 参数就是在前面初始化 Channel 所传入的 SelectionKey.OPACCEPT 事件,所以 OPACCEPT 事件会被注册到 Channel 的事件集合中。

11.监听Accept事件:16

java @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,则监听readInterestOp if ((interestOps & readInterestOp) == 0) { //NioServerSocketChannel: readInterestOp = OP_ACCEPT = 1 << 4 = 16 logger.info("interest ops: " + readInterestOp); selectionKey.interestOps(interestOps | readInterestOp); } }

到此为止,整个服务端已经真正启动完毕。我们总结一下服务端启动的全流程,如下图所示。

图片5.png

创建服务端 Channel:本质是创建 JDK 底层原生的 Channel,并初始化几个重要的属性,包括 id、unsafe、pipeline 等。

初始化服务端 Channel:设置 Socket 参数以及用户自定义属性,并添加1个特殊的处理器 ChannelInitializer,ChannelInitializer的功能是添加 LoggingHandler 和 ServerBootstrapAcceptor,但是并没有添加进去。

注册服务端 Channel:调用 JDK 底层将 Channel 注册到 Selector上。执行ChannelInitializer的initChannel真正添加handler。

端口绑定:调用 JDK 底层进行端口绑定,并触发 channelActive 事件,把 OP_ACCEPT 事件注册到NioServerSocketChannel 的事件集合中。

总结

```java •启动服务的本质: ​ Selector selector = sun.nio.ch.SelectorProviderImpl.openSelector()

ServerSocketChannel serverSocketChannel = provider.openServerSocketChannel()

selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

javaChannel().bind(localAddress, config.getBacklog());

selectionKey.interestOps(OP_ACCEPT);

知识点: •Selector 是在 new NioEventLoopGroup()(创建一批 NioEventLoop)时创建。 ​ •第一次 Register 并不是监听 OPACCEPT,而是 0: ​ selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this) 。 ​ •最终监听 OPACCEPT 是通过 bind 完成后的 fireChannelActive() 来触发的。 ​ •NioEventLoop 是通过 Register 操作的执行来完成启动的。 ​ •类似 ChannelInitializer,一些 Hander 可以设计成一次性的,用完就移除,例如授权 ​ ```

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/891926.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

如何利用 ChatGPT 进行自动数据清理和预处理

推荐&#xff1a;使用 NSDT场景编辑器助你快速搭建可二次编辑的3D应用场景 ChatGPT 已经成为一把可用于多种应用的瑞士军刀&#xff0c;并且有大量的空间将 ChatGPT 集成到数据科学工作流程中。 如果您曾经在真实数据集上训练过机器学习模型&#xff0c;您就会知道数据清理和预…

Linux debian12解压和压缩.rar文件教程

一、Debian12安装rar命令 sudo apt install rar二、使用rar软件 1.解压文件 命令格式&#xff1a; rar x 文件名.rar实力测试&#xff1a; [rootdoudou tmp]# rar x test.rar2.压缩文件 test是一个文件夹 命令格式&#xff1a; rar a 文件名.rar 文件夹名实例测试&#x…

Java的AQS框架是如何支撑起整个并发库的

如何设计一个抽象队列同步器 引言AQS需要解决哪些场景下的问题互斥模式获取锁抢锁失败入队 释放锁小总结 共享模式获取共享资源释放共享资源唤醒丢失问题 小总结 混合模式获取写锁释放写锁获取读锁读锁是否应该阻塞 释放读锁小总结 栅栏模式等待递减计数 条件变量模式等待条件成…

如何将图片应用于所有的PPT页面?

问题&#xff1a;如何快速将图片应用到所有PPT页面&#xff1f; 解答&#xff1a;有两种方法可以解决这个问题。第一种用母板。第二种用PPT背景功能。 解决有时候汇报的时候&#xff0c;ppt中背景图片修改不了以及不知道如何查找&#xff0c;今天按照逆向过程进行操作 方法1…

Intelij IDEA 配置Tomcat解决Application Server不显示的问题

今天搭建war工程时部署项目发现&#xff0c;IDEA的控制台没有Application Servers&#xff0c;在网上查了一下&#xff0c;总结几个比较好的解决方法&#xff0c;为了方便自己和其他人以后碰到相同的问题&#xff0c;不再浪费时间再次寻找解决办法。 Intelij IDEA 配置Tomcat时…

PyMuPDF`库实现PDF旋转功能

本文介绍了一个简单的Python应用程序&#xff0c;用于将PDF文件转换为旋转90度的PDF文件。主要用于csdn网站中导出的博客pdf是横向的&#xff0c;看起来不是很方便&#xff0c;才想到用python编制一个将pdf从横向转为纵向的功能。 功能 该PDF转换工具具有以下功能&#xff1a…

pdf转word最简单方法~

pdf转word最简单方法&#xff01;pdf转word最简单方法我们都知道&#xff0c;PDF文件是一种只读文件格式&#xff0c;无法按照需求对PDF文件进行更改与编辑&#xff0c;从而影响到了PDF文件的使用。所以&#xff0c;我们需要将PDF文件转换为word文档&#xff0c;以此来保证文件…

js 小程序限流函数 return闭包函数执行不了

问题&#xff1a; 调用限流 &#xff0c;没走闭包的函数&#xff1a; checkBalanceReq&#xff08;&#xff09; loadsh.js // 限流 const throttle (fn, context, interval) > {console.log(">>>>cmm throttle", context, interval)let canRun…

Webgl 存储限定符attribute、gl.getAttribLocation、gl.vertexAttrib3f及其同族函数和矢量版本的介绍

目录 attribute变量规范 获取attribute变量的存储位置 gl.getAttribLocation&#xff08;&#xff09;函数的规范&#xff1a; 向attribute变量赋值 gl.vertexAttrib3f&#xff08;&#xff09;的规范。 gl.vertexAttrib3f&#xff08;&#xff09;的同族函数 示例代码…

达梦数据库8用户管理以及忘记sysdba密码修改办法

达梦数据库8用户管理&达梦数据库v8忘记sysdba密码&#xff0c;修改办法。 达梦数据库8用户管理1.创建用户的语法:2.锁定/解锁用户3.修改用户的密码&#xff08;同样要符合密码策略PWD_POLICY&#xff09;4.修改用户默认表空间5.删除用户6.同样地可以使用DM管理工具进行创建…

基于docker搭建pytest自动化测试环境(docker+pytest+jenkins+allure)

pytest搭建自动化测试环境&#xff08;dockerpytestjenkinsallure&#xff09; 这里我以ubuntu18为例 如果有docker环境&#xff0c;可以直接拉取我打包好的镜像docker pull ziyigun/jenkins:v1.0 1 搭建Docker 1.1 安装docker # 配置docker安装环境 sudo apt-get install ap…

js 闭包和 垃圾回收机制

js 闭包和 垃圾回收机制 闭包里面的变量 是局部变量 但是 由于一直使用&#xff0c;引用&#xff0c;没有被回收 所以相当于 全局变量

Docker案例分析:创建并运行一个Redis容器

步骤一&#xff1a;到DockerHub搜索Redis镜像 步骤二&#xff1a;查看Redis镜像文档中的帮助信息 步骤三&#xff1a;利用docker run命令运行一个Redis容器 可以看到第二行的命令是支持数据持久化的创建容器命令&#xff0c;但是我们使用的命令如下&#xff1a; docker run -…

Vue3 使用json编辑器

安装 npm install json-editor-vue3 main中引入 main.js 中加入下面代码 import "jsoneditor";不然会有报错&#xff0c;如jsoneditor does not provide an export named ‘default’。 图片信息来源-github 代码示例 <template><json-editor-vue class…

ssm+vue校园美食交流系统源码

ssmvue校园美食交流系统源码和论文026 开发工具&#xff1a;idea 数据库mysql5.7 数据库链接工具&#xff1a;navcat,小海豚等 技术&#xff1a;ssm 摘 要 随着现在网络的快速发展&#xff0c;网上管理系统也逐渐快速发展起来&#xff0c;网上管理模式很快融入到了许多商…

Stable Diffusion入门修炼手册

简介 作为新入门的新手&#xff0c;通常安装完Stable Diffusion之后&#xff0c;一打开界面&#xff0c;在文生图输入girl或者dog&#xff0c;结果出来的画面比较糟糕&#xff0c;看起来像素很低&#xff0c;画面不清晰&#xff0c;人物也不怎么美&#xff0c;等等其他问题&am…

tomcat中的BIO与NIO发展

tomcat中的NIO发展 前言 Tomcat目前支持BIO&#xff08;阻塞 I/O&#xff09;、NIO&#xff08;非阻塞 I/O&#xff09;、AIO&#xff08;异步非阻塞式IO&#xff0c;NIO的升级版&#xff09;、APR&#xff08;Apache可移植运行库&#xff09;模型&#xff0c;本文主要介绍NI…

如何使用CSS实现一个平滑过渡效果?

聚沙成塔每天进步一点点 ⭐ 专栏简介⭐ 使用CSS实现平滑过渡效果⭐ 写在最后 ⭐ 专栏简介 前端入门之旅&#xff1a;探索Web开发的奇妙世界 记得点击上方或者右侧链接订阅本专栏哦 几何带你启航前端之旅 欢迎来到前端入门之旅&#xff01;这个专栏是为那些对Web开发感兴趣、刚…

【FAQ】调用视频汇聚平台EasyCVR的iframe地址,视频无法播放的原因排查

有用户反馈&#xff0c;在调用iframe地址后嵌入用户自己的前端页面&#xff0c;视频无法播放并且要求登录。 安防监控视频汇聚平台EasyCVR基于云边端一体化架构&#xff0c;具有强大的数据接入、处理及分发能力&#xff0c;可提供视频监控直播、云端录像、视频云存储、视频集中…

nginx反向代理流程

一、nginx反向代理流程 反向代理&#xff1a;使用代理服务器来接受internet上的连接请求&#xff0c;然后将请求转发给内部网络中的上游服务器&#xff0c;并将上游服务器得到的结果返回给请求连接的客户端&#xff0c;代理服务器对外表现就是一个web服务器。Nginx就经常拿来做…