关闭

Netty 服务端启动过程

时间: 2019-01-24阅读: 702标签: Netty

在 Netty 中创建 1 个 NioServerSocketChannel 在指定的端口监听客户端连接,这个过程主要有以下  个步骤:

  1. 创建 NioServerSocketChannel
  2. 初始化并注册 NioServerSocketChannel
  3. 绑定指定端口


首先列出一个简易服务端的启动代码:  

public void start() {
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap sbs = new ServerBootstrap()
                //添加 group
                .group(bossGroup, workerGroup)
                //指定服务端 Channel 类型
                .channel(NioServerSocketChannel.class)
                //添加服务端 Channel 的 Handler
                .handler(new HelloWorldServerHandler())
                //添加客户端 Channel 的 Handler
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        //为后续接入的客户端 Channel 准备的字符串编解码 Handler 
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new StringEncoder());
                    }
                });
        //监听指定的端口
        ChannelFuture future = sbs.bind(port).sync();
        System.out.println("Server start listen at " + port);
        future.channel().closeFuture().sync();
    } catch (Exception e) {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

下面就从 ServerBootstrap 的 bind(int port)方法开始分析服务端的 NioServerSocketChannel 的创建过程。 


1. 创建 NioServerSocketChannel 

跟随 bind 方法的调用,最终在 AbstractBootstrap 类的 doBind()方法找到了初始化,注册和绑定方法调用: 

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 {
        //....
    }
}


2.初始化并注册 NioServerSocketChannel

首先来看一下这个 initAndRegister()方法:

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        //创建 Channel
        channel = channelFactory.newChannel();
        //初始化 Channel
        init(channel);
    } catch (Throwable t) {
        //...
    }

    //注册
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    //...
}


Channel 也是通过工厂类来创建的,这个工厂默认是 ReflectiveChannelFactory,是在前面启动代码中,设置服务端 Channel 类型时创建的。通过名字可以知道,是用反射的方式创建了 Channel 对象。

init()方法有两种实现,这里分析的是 ServerBootstrap 的实现:

@Override
void init(Channel channel) throws Exception {
    //... option 的设置省略掉
    //pipeline 的创建,默认使用的 DefaultPipeline
    ChannelPipeline p = channel.pipeline();

    //... 客户端 Channel 相关配置的保存

    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            //这里添加的是启动代码中,服务端的 Handler
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }

            // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
            // In this case the initChannel(...) method will only be called after this method returns. Because
            // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
            // placed in front of the ServerBootstrapAcceptor.
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    //这里添加了一个 Accepter,用来处理新连接的接入
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}


初始化 Channel 这个动作,主要做了 4 件事:

  1. 创建 pipeline
  2. 为 Channel 添加用户创建的 Handler 
  3. 添加 Accepter
  4. 其他属性的设置

接下来分析 Channel 的注册,需要关注的是这行代码:

ChannelFuture regFuture = config().group().register(channel);


config()方法获取了启动时创建的 config 对象,这个对象的 group()方法就返回了启动时传入的 bossGroup。启动代码中传入了两个 group,返回的为什么是 boosGroup 呢?查看启动代码中的 group(EventLoopGroup parentGroup, EventLoopGroup childGroup)方法,在它第一行就调用了 super.group(parentGroup),将第一个 group 对象传给了父类 AbstractBootstrap。而此处 config 调用的 group()方法返回的正是父类中的 group。

因为这里是一个 NioEventLoopGroup 对象,所以使用的 register(channel)方法是 MultithreadEventLoopGroup 中的。

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


查看 next()方法可以发现,最终是调用之前创建 group 时创建的 chooser 的 next()方法,该方法会返回一个 NioEventLooop 对象(EventLoop 是在这里分配的),它的 register()方法是在父类 SingleThreadEventLoop 中实现的。最终调用了 AbstractChannel 中的注册方法。 

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    //...
    //将前面返回的 eventLoop 保存起来
    AbstractChannel.this.eventLoop = eventLoop;
    //判断 eventLoop 中的 thread 是否是当前线程
    //初次启动时,eventLoop 中的 thread 为 null
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            //将注册任务传进去
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    //注册
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            //...
        }
    }
}


将注册动作封装成一个任务,然后交给 eventLoop 对象处理。

@Override
public void execute(Runnable task) {
    //...
    //这里通用是 false
    boolean inEventLoop = inEventLoop();
    if (inEventLoop) {
        addTask(task);
    } else {
        //启动线程
        startThread();
        addTask(task);//将前面传进来的注册任务添加进队列
        if (isShutdown() && removeTask(task)) {
            reject();
        }
    }

    if (!addTaskWakesUp && wakesUpForTask(task)) {
        wakeup(inEventLoop);
    }
}

private void startThread() {
    //判断是否需要启动线程
    if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
        if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
            //启动线程
            doStartThread();
        }
    }
}


上面代码中的 startThread()方法有个 STATE_UPDATER,它是用来更新该对象的 state 属性,是一个线程安全的操作。state 默认值为 ST_NOT_STARTED,所以第一次进入该方法,条件判断为 true,接下来进行 CAS 操作,将 state 设置为 ST_STARTED,然后调用 doStartThread()方法。当 group 中的线程都启用之后,下一次 chooser 再选中这个线程,startThread()方法中的第一个 if 的条件判断就是 false 了,不会再创建新的线程。 

private void doStartThread() {
    assert thread == null;
    //这个 executor 就是构建 group 时,创建出来的 executor
    executor.execute(new Runnable() {
        @Override
        public void run() {
            thread = Thread.currentThread();
            if (interrupted) {
                thread.interrupt();
            }

            boolean success = false;
            updateLastExecutionTime();
            try {
                //前面创建的是 NioEventLoop
                SingleThreadEventExecutor.this.run();
                success = true;
            } catch (Throwable t) {
                logger.warn("Unexpected exception from an event executor: ", t);
            } finally {
                for (;;) {
                    //更新 state
                    int oldState = STATE_UPDATER.get(SingleThreadEventExecutor.this);
                    if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
                            SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
                        break;
                    }
                }
                //...
            }
        }
    });
}


 前一篇分析 EventLoopGroup 创建时说过,会在 EventLoop 保存一个 executor 对象的引用,最终个任务就是交给这个 executor 来处理的。executor 的 execute(Runnable task) 方法会创建新线程,并执行传入的 task。接下来看一下 NioEventLoop 中的 run() 方法。

protected void run() {
    for (;;) {
        try {
            //计算 select 策略,当前有任务时,会进行一次 selectNow(NIO),返回就绪的 key 个数
            //显然 switch 中没有匹配项,直接跳出 switch
            //无任务时,则直接返回 SelectStrategy.SELECT
            //这里的 SelectStrategy.CONTINUE 感觉不会匹配到
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    //当没有可处理的任务时,直接进行 select 操作
                    // wakenUp.getAndSet(false) 返回的是 oldValue,由于默认值是 false
                    // 所以第一次返回的是 false
                    select(wakenUp.getAndSet(false));

                    // 'wakenUp.compareAndSet(false, true)' is always evaluated
                    // before calling 'selector.wakeup()' to reduce the wake-up
                    // overhead. (Selector.wakeup() is an expensive operation.)
                    //
                    // However, there is a race condition in this approach.
                    // The race condition is triggered when 'wakenUp' is set to
                    // true too early.
                    //
                    // 'wakenUp' is set to true too early if:
                    // 1) Selector is waken up between 'wakenUp.set(false)' and
                    //    'selector.select(...)'. (BAD)
                    // 2) Selector is waken up between 'selector.select(...)' and
                    //    'if (wakenUp.get()) { ... }'. (OK)
                    //
                    // In the first case, 'wakenUp' is set to true and the
                    // following 'selector.select(...)' will wake up immediately.
                    // Until 'wakenUp' is set to false again in the next round,
                    // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
                    // any attempt to wake up the Selector will fail, too, causing
                    // the following 'selector.select(...)' call to block
                    // unnecessarily.
                    //
                    // To fix this problem, we wake up the selector again if wakenUp
                    // is true immediately after selector.select(...).
                    // It is inefficient in that it wakes up the selector for both
                    // the first case (BAD - wake-up required) and the second case
                    // (OK - no wake-up required).

                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
                    // fallthrough
            }

            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            //根据比例来处理 IO 事件和任务
            if (ioRatio == 100) {
                try {
                    //处理就绪的 key
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    //执行任务
                    runAllTasks();
                }
            } else {
                final long ioStartTime = System.nanoTime();
                try {
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    // 计算出处理 IO 事件的时间,然后根据比例算出执行任务的时间
                    final long ioTime = System.nanoTime() - ioStartTime;
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        // Always handle shutdown even if the loop processing threw an exception.
        try {
            if (isShuttingDown()) {
                closeAll();
                if (confirmShutdown()) {
                    return;
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
    }
}


run()方法主要是做 select 操作,和处理 IO 事件和任务队列中的任务,这部分内容下一篇文章再分析。从 executor 执行 execute()方法开始,由 Netyy 管理的线程就开始启动运行了。实际上此时的 NioServerSocketChannel 对象还没有注册到 Netty 线程的 Selector 上,Debug 结果如下图:  


上图中的 startThread()方法实际上是给 executor 提交了一个任务,紧接着 main 线程就调用了 addTask()方法,将 task 添加到 EventLoop 对象的任务队列中,而这个 task 的内容就是执行注册操作。在添加了注册任务之后,Netty 线程就会在 select 完成后,执行队列中的任务,将 NioServerSocketChannel 注册到该线程的 Selector 上。接下来分析一下 AbstractChannel 的 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.
        //添加服务端 Channel 的 Handler
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        //触发通道注册事件在 pipeline 上传播
        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()) {//第一次运行到这儿时,结果为 false,因为此时还没有 bind
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                // This channel was registered before and autoRead() is set. This means we need to begin read
                // again so that we process inbound data.
                //
                // See https://github.com/netty/netty/issues/4805
                beginRead();
            }
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}


doRegister()方法实际上就是 Java NIO 中将通道注册到 Selector 上的操作:  

selectionKey = javaChannel().register(eventLoop().selector, 0, this);//这里感兴趣的事件传入的是 0

pipeline.invokeHandlerAddedIfNeeded() 和 pipeline.fireChannelRegistered() 则是用来添加 Handler 并触发 Handler 别添加的事件的动作。

在 isActive()这个方法,由于当前是 NioServerSocketChannel,所以实际上是判断当前通道是否成功绑定到一个地址,很显然到目前为止,只是创建了通道并注册到 Selector 上,还没由绑定。


3. 绑定指定端口

在 initAndRegister()方法结束后,main 线程开始调用 doBind0()方法,该方法将绑定操作封装成任务交给 Netty 线程去执行。最后,调用 DefaultPipeline 中的 HeadContext 的 bind()方法,然后通过 unsafe.bind(localAddress,promise)完成绑定:

@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    //...
    //显然这里返回的是 false
    boolean wasActive = isActive();
    try {
        //绑定操作
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }

    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                //这里才是触发服务端 Channel 激活事件的地方
                pipeline.fireChannelActive();
            }
        });
    }

    safeSetSuccess(promise);
}


这个过程,建议 Debug 跟一下代码,比较清楚代码是如何一步一步到 HeadContext 中来的。接下来分析一下 doBind()方法:  

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


最终是根据平台及其 Java 版本来调用 JDK 中的绑定方法。在绑定完成后,会触发通道激活事件,在 HeadContext 中经过时,发现它里面有这么一行代码: 

readIfIsAutoRead();


Debug 一下,发现这个方法最终会调用到 HeadContext 的 read()方法,该方法是调用了 unsafe.beginRead(),紧接着就到了 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();
    if ((interestOps & readInterestOp) == 0) {//说明对 OP_ACCEPT 不感兴趣
        selectionKey.interestOps(interestOps | readInterestOp);//通过 | 修改感兴趣的事件
    }
}

前面通过反射创建 NioServerSocketChannel 对象时,调用了父类也就是 AbstractNioChannel 的构造方法,将 readInterestOp 设置为 16 了,在 NIO 中就是 OP_ACCEPT。从此,该 NioServerSocketChannel 就可以接收客户端连接了。


4. 总结

在 Netty 服务端启动过程中,主线程仅仅是创建了 EventLoopGroup 和启动引导对象,然后发起绑定操作。这个过程中的绑定,注册等操作都是主线程封装成任务交给 Netty 线程去执行的。

由于 Netty 代码中抽象类和接口都比较多,所以某些地方调用的方法有很多种实现,不熟悉的时候可以通过 Debug 来确定。 

来自:https://www.cnblogs.com/magexi/p/10228780.html


站长推荐

1.云服务推荐: 国内主流云服务商,各类云产品的最新活动,优惠券领取。地址:阿里云腾讯云华为云

2.广告联盟: 整理了目前主流的广告联盟平台,如果你有流量,可以作为参考选择适合你的平台点击进入

链接: http://www.fly63.com/article/detial/1879

关闭

内容以共享、参考、研究为目的,不存在任何商业目的。其版权属原作者所有,如有侵权或违规,请与小编联系!情况属实本人将予以删除!