netty源码阅读之客户端新连接之分配线程并注册selector

谁借莪１个温暖的怀抱￠ 2022-05-17 13:18 136阅读 0赞

首先，我们看看服务端channel的pipeline里面的构成：

Head---->ServerBootstrapAcceptor--->Tail。

也就是，新连接进来之后，会经过Head，这个Head就是我们之前这篇文章《[netty源码阅读之服务器启动之端口绑定][netty]》的HeadContext，绑定端口。是不是串起来了，之前一个不知道为什么用HeadContext的实现，现在知道了吧？关于HeadContext，在后面介绍pipeline的文章会继续讲解。

然后传播到ServerBootstrapAcceptor，这个跟Head和Tail，其实都是Handler。这篇《[netty源码阅读之服务器启动之服务端channel的初始化][netty_channel]》稍微提到过ServerBootstrapAcceptor（服务端启动的时候，默认会添加这个Handler，用来处理新连接的逻辑），现在我们详细分析它的channelRead怎么处理新请求：

1、添加childHandler（到pipeline里面）

2、设置options和attrs

3、选择NioEventLoop并注册selector事件

接着这一篇文章《[netty源码阅读之客户端新连接之检测新连接][netty 1]》，检测到新链接之后，看源码：

@Override
            public void read() {
                assert eventLoop().inEventLoop();
                final ChannelConfig config = config();
                final ChannelPipeline pipeline = pipeline();
                final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
                allocHandle.reset(config);
    
                boolean closed = false;
                Throwable exception = null;
                try {
                    try {
                        do {
                            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;
                        pipeline.fireChannelRead(readBuf.get(i));
                    }
                    ...
                } finally {
                    ...
                }
            }

后面从pipeline.fireChannelRead(readBuf.get(i))一步步进入(这个readBuf呢，是我们之前每次接受连接，都会放到readBuf里面)，最后会调用到ServerBootstrapAcceptor的channelRead方法：

public void channelRead(ChannelHandlerContext ctx, Object msg) {
                final Channel child = (Channel) msg;
    
                child.pipeline().addLast(childHandler);
    
                for (Entry<ChannelOption<?>, Object> e: childOptions) {
                    try {
                        if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
                            logger.warn("Unknown channel option: " + e);
                        }
                    } catch (Throwable t) {
                        logger.warn("Failed to set a channel option: " + child, t);
                    }
                }
    
                for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                    child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
                }
    
                try {
                    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);
                }
            }

这里面，就包含了上面的所有过程。

### 添加childHandler ###

这个childHandler，一步步追踪，会发现就是用户这段代码进来的：

.childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) {
                                ch.pipeline().addLast(new AuthHandler());
                                //..
    
                            }
                        });

而这个ChannelInitializer又是客户端channel特殊的channel，看上面的代码，通过这个类的initChannel方法，我们可以得到SocketChannel，并进一步得到pipeline，然后就可以通过pipeline，进一步添加我们真正想要handler，为什么要这样设计？

我们继续看ChannelInitializer的设计吧。在服务端添加这个类的时候，也就是child.pipeline().addLast(childHandler);这一步的add的时候，一步步跟进去，会发现他调用到ChannelInitializer的handlerAdded方法：

@Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            if (ctx.channel().isRegistered()) {
                // This should always be true with our current DefaultChannelPipeline implementation.
                // The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering
                // suprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers
                // will be added in the expected order.
                initChannel(ctx);
            }
        }

然后这个方法调用它的initChannel方法：

@SuppressWarnings("unchecked")
        private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
            if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // 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 {
                    remove(ctx);
                }
                return true;
            }
            return false;
        }

initChannel就会回调到用户匿名添加的initChannel方法，也就是:

@Override
                            public void initChannel(SocketChannel ch) {
                                ch.pipeline().addLast(new AuthHandler());
                                //..
    
                            }

所以就会把后面的Handler添加进去。

最后remove自己，关于remove，后面文章我们会继续探讨。

### 设置options和attrs ###

接下去添加options

for (Entry<ChannelOption<?>, Object> e: childOptions) {
                    try {
                        if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
                            logger.warn("Unknown channel option: " + e);
                        }
                    } catch (Throwable t) {
                        logger.warn("Failed to set a channel option: " + child, t);
                    }
                }

我们在用户代码设置了这个东西：.childOption(ChannelOption.TCP\_NODELAY, true)，会变成childOptions传入这里来（childOptions是和TCP读写有关的一些参数）。

child.config()返回的就是在这篇文章《[netty源码阅读之客户端新连接之创建NioSocketChannel][netty_NioSocketChannel]》创建的new NioSocketChannelConfig()，在之前我们讲过，这个config就是设置tcp连接的一些参数，在这里得到了验证。

继续看：

for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                    child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
                }

这样attrs就是用户代码：.childAttr(AttributeKey.newInstance("childAttr"), "childAttrValue")这个地方设置的任意的属性，childAttrs的一些作用就是可以在客户端上面绑定一些属性，例如秘钥等。

### 选择NioEventLoop并注册selector ###

最后从这段代码的register进去：

childGroup.register(child).addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (!future.isSuccess()) {
                                forceClose(child, future.cause());
                            }
                        }
                    });

又回到了类似服务端注册的代码，因为这个是一个group(workGroup)，所以首先进入MultithreadEventLoopGroup的register：

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

然后我们看这个next，不停地进入：

@Override
        public EventExecutor next() {
            return chooser.next();
        }

这个chooser是不是很熟悉，没错，它就是我们这篇文章《[netty源码阅读之NioEventLoop之NioEventLoop创建][netty_NioEventLoop_NioEventLoop]》里面创建的，经过优化的chooser！这里又用上了chooser，所以，它就是调用优化的算法，返回一个NioEventLoop。

从刚刚的next之后的register继续深入，（调用了ByteUnsafe的register）又来到这段熟悉的代码：

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

把刚刚返回的nioEventLoop绑定到这个客户端Channel上面。

由于这里的发起线程是服务端channel的NioEventLoop，所以inEventLoop为false（不在当前客户端的线程里面），就要eventLoop就要新发起一个任务。从register0()方法不停地进入，就是段代码：

@Override
        protected void doRegister() throws Exception {
            boolean selected = false;
            for (;;) {
                try {
                    selectionKey = javaChannel().register(eventLoop().selector, 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;
                    }
                }
            }
        }

在这里，就是刚刚的nioEventLoop的selector注册进去一个0事件进去（也就是什么都不注册），并且把this，也就是客户端channel作为attachment传进去。

这里需要留意的是注册了0事件进去，会和我们下一篇文章相关。

我们这里再回顾两点：

1、nioEventLoop的selector就是在这篇文章《[netty源码阅读之NioEventLoop之NioEventLoop创建][netty_NioEventLoop_NioEventLoop]》添加进去的（创建NioEventLoop的时候，顺便创建selector）。

2、想想attachment什么地方用到了？

《[netty源码阅读之NioEventLoop之NioEventLoop执行-----processSelectedKey()执行][netty_NioEventLoop_NioEventLoop_-----processSelectedKey]》这篇文章这里，把这个attachment也就是channel找出来，然后在这篇文章《[netty源码阅读之客户端新连接之检测新连接][netty 1]》，使用channel的unsafe读取数据！

[netty]: https://blog.csdn.net/fst438060684/article/details/81431560
[netty_channel]: https://blog.csdn.net/fst438060684/article/details/81428985
[netty 1]: https://blog.csdn.net/fst438060684/article/details/81585914
[netty_NioSocketChannel]: https://blog.csdn.net/fst438060684/article/details/81586156
[netty_NioEventLoop_NioEventLoop]: https://blog.csdn.net/fst438060684/article/details/81534359
[netty_NioEventLoop_NioEventLoop_-----processSelectedKey]: https://blog.csdn.net/fst438060684/article/details/81559977