Netty分布式NioSocketChannel注册到selector方法的示例分析

这篇文章给大家分享的是有关Netty分布式NIOSocketChannel注册到selector方法的示例分析的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。

我们回到最初的NioMessageUnsafe的read()方法:

public void read() {    //必须是NioEventLoop方法调用的, 不能通过外部线程调用    assert eventLoop().inEventLoop();    //服务端channel的config    final ChannelConfig config = config();    //服务端channel的pipeline    final ChannelPipeline pipeline = pipeline();    //处理服务端接入的速率    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();    //设置配置    allocHandle.reset(config);    boolean closed = false;    Throwable exception = null;    try {        try {            do {                //创建jdk底层的channel                //readBuf用于临时承载读到链接                int localRead = doReadMessages(readBuf);                if (localRead == 0) {                    break;                }                if (localRead < 0) {                    closed = true;                    break;                }                //分配器将读到的链接进行计数                allocHandle.incMessagesRead(localRead);                //连接数是否超过最大值            } while (allocHandle.continueReading());        } catch (Throwable t) {            exception = t;        }        int size = readBuf.size();        //遍历每一条客户端连接        for (int i = 0; i < size; i ++) {            readPending = false;            //传递事件, 将创建NiOSokectChannel进行传递            //最终会调用ServerBootstrap的内部类ServerBootstrapAcceptor的channelRead()方法            pipeline.fireChannelRead(readBuf.get(i));        }        readBuf.clear();        allocHandle.readComplete();        pipeline.fireChannelReadComplete();        //代码省略    } finally {        //代码省略    }}

在while循环结束之后, 将会通过一个for循环遍历readBuf集合, 并将创建的NioSocketChannel传入fireChannelRead()中, 传播channel的读取事件

有关pipeline的知识, 我们下一章会详细剖析, 并会根据剖析后的内容回顾之前的有关pipeline的操作, 这里我们只需知道, 通过fireChannelRead()我们最终调用了ServerBootstrap的内部类ServerBootstrapAcceptor 中的channelRead()方法

跟到channelRead()方法中:

public void channelRead(ChannelHandlerContext ctx, Object msg) {    final Channel child = (Channel) msg;    //代码省略    try {        //work线程注册channel        childGroup.reGISter(child).addListener(new ChannelFutureListener() {            @Override            public void operationComplete(ChannelFuture future) throws Exception {                if (!future.isSuccess()) {                    forceClose(child, future.cause());                }            }        });    } catch (Throwable t) {        forceClose(child, t);    }}

其中参数的msg就是最初传入fireChannelRead()方法的NioSocketChannel

所以这里可以通过 final Channel child = (Channel) msg 这种方式拿到NioSocketChannel

其中childGroup是我们最初初始化的work线程, 这里的register()方法跟boss线程一样, 通过next()方法获选择一个线程进行注册, 这里不再赘述

我们紧跟调用链, 跟到SingleThreadEventLoop的register()方法:

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

这里的unsafe(), 根据我们之前的剖析, 是NioByteUnsafe, 这里的register最终会调用AbstractUnsafe的register()方法, 并NioSocketChannel

不知道同学们是否记得, 当初NioServerSocketChannel注册的时候也走的这个方法

我们跟到register()这个方法中:

public final void register(EventLoop eventLoop, final ChannelPromise promise) {    //省略验证代码    //所有的复制操作, 都交给eventLoop处理    AbstractChannel.this.eventLoop = eventLoop;    if (eventLoop.inEventLoop()) {        //做实际主注册        register0(promise);    } else {        try {            eventLoop.execute(new Runnable() {                @Override                public void run() {                    register0(promise);                }            });        } catch (Throwable t) {            //代码省略        }    }}

我们学习过NioEventLoop相关知识之后, 应该对这部分代码不太陌生, 首先判断是不是当前NioEventLoop线程, 如果是, 则直接进行注册操作, 如果不是, 则封装成task在当前NioEventLoop中执行

走到这里不难明白, 这里并不是当前NioEventLoop线程, 这是boss线程执行的, 所以这里会走到else, 如果是第一次的连接操作, work线程的NioEventLoop并没有启动, 所以这里也会启动NioEventLoop, 并开始轮询操作

跟到register0(promise)中看其是如何做实际操作的:

private void register0(ChannelPromise promise) {    try {        //省略代码        //做实际的注册        doRegister();        neverRegistered = false;        registered = true;        //触发事件        pipeline.invokeHandlerAddedIfNeeded();        safeSetSuccess(promise);        //触发注册成功事件        pipeline.fireChannelRegistered();        if (isActive()) {            if (firstRegistration) {                //传播active事件(4)                pipeline.fireChannelActive();            } else if (config().isAutoRead()) {                beginRead();            }        }    } catch (Throwable t) {        //省略代码    }}

这段代码我们同样并不陌生, 因为NioServerSokectChannel中也走这一部分, 我们继续关注doRegister()方法:

protected void doRegister() throws Exception {    boolean selected = false;    for (;;) {        try {            //jdk底层的注册方法            //第一个参数为selector, 第二个参数表示不关心任何事件            selectionKey = javaChannel().register(eventLoop().selector, 0, this);            return;        } catch (CancelledKeyException e) {            //省略代码        }    }}

这部分也是我们之前剖析过的jdk底层的注册, 只是不同的是, 这里的javaChannel()是SocketChanel而不是ServerSocketChannel

同样, 这里也是表示不关心任何事件, 只是在当前NioEventLoop绑定的selector上注册

感谢各位的阅读！关于“Netty分布式NioSocketChannel注册到selector方法的示例分析”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！