Netty源码分析(3)-新连接接入

发表于 | 分类于
字数统计: 1.8k

上文介绍了服务端启动的过程，这篇文章介绍接收新连接的过程。

(BOSS)EventLoop的启动过程

上回书说到在注册channel过程中有这样的一段代码，当时说是BossEventLoop reactor线程启动的入口，我们从这里去分析本文。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    //...
    AbstractChannel.this.eventLoop = eventLoop;
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        eventLoop.execute(new Runnable() {
            @Override
            public void run() {
                register0(promise);
            }
        });
    }
}

这段代码就是将channel的注册行为抽象成了一次异步的任务，而eventLoop在执行execute方法时候也同样启动了。我们这里的channel.eventLoop指的是当前channel对应的单个EventLoop(我们这里分析的是BossEventLoop)，而SingleThreadEventExecutor是EventLoop的父类，所以我们根据eventLoop的execute方法追溯到了这里

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//SingleThreadEventExecutor.class
@Override
public void execute(Runnable task) {
   //...
    boolean inEventLoop = inEventLoop();
    if (inEventLoop) {
        addTask(task);
    } else {
        startThread();
        addTask(task);
        if (isShutdown() && removeTask(task)) {
            reject();
        }
    }
    //...
}

简单地调用栈是这样的startThread()->doStartThread()->run()->NioEventLoop#run
NioEventLoop的run方法是一个很重要的方法详情见Netty源码分析(2)-Reactor线程模型，这是NioEventLoop启动的地方，在这里进行无限循环。执行NioEventLoop的三板斧：轮询注册在selector上的IO事件、处理IO事件、同步执行异步task。而新连接的接入过程就位于第二步骤（处理IO事件），我们接下来去分析。

(BOSS)EventLoop#processSelectedKeys()

从无参的processSelectedKeys()追到了processSelectedKeys(SelectionKey k, AbstractNioChannel ch)中

1
2
3
4
5
6
7
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    //...
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
        unsafe.read();
    }
    //...
}

这里有一个疑问，和当前channel绑定的eventLoop是如何确定的？
答：在channel注册过程中，通过next()方法从bossGroup中随机选择的一个eventLoop

我们看到此时当EventLoop（BossGroup中的一员）轮询到了ACCEPT事件会调用unsafe.read()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    do {
        int localRead = doReadMessages(readBuf);
        if (localRead == 0) {
            break;
        }
        if (localRead < 0) {
            closed = true;
            break;
        }
    } while (allocHandle.continueReading());
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    pipeline.fireChannelReadComplete();
}

这里重点关注doReadMessages和fireChannelRead方法

doReadMessages

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
//NioServerSocketChannel.class
protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = javaChannel().accept();
    try {
        if (ch != null) {
            buf.add(new NioSocketChannel(this, ch));
            return 1;
        }
    } catch (Throwable t) {
        logger.warn("Failed to create a new channel from an accepted socket.", t);
        try {
            ch.close();
        } catch (Throwable t2) {
            logger.warn("Failed to close a socket.", t2);
        }
    }
    return 0;
}

第一行代码就是jdk nio接收新连接的代码，因为是监听到了ACCEPT事件所以这里会立刻返回一个SocketChannel，这个也是nio中的channle。我们可以看到buf是一个list容器，再将socketChannel放到容器之前，还要对其进行包装（包装为netty中的channel类型）。
下面去看看具体咋包装的

1
2
3
4
public NioSocketChannel(Channel parent, SocketChannel socket) {
    super(parent, socket);
    config = new NioSocketChannelConfig(this, socket.socket());
}

诶等一等，这个代码看着有点眼熟啊，这就是Netty源码分析(1)-服务端启动过程中new Channel那段代码，所以就不去看了，大概创建了这样几个东西

  • channel,
  • channelConfig,
  • 保存了变量到成员变量,
  • channelId,unsafe,pipeline

不过这里还是有一些不同的，这里注册的事件是SelectionKey.OP_READ，而服务端channel注册的是SelectionKey.OP_ACCEPT事件；而且这里的channel是NioScoketChannel

fireChannelRead

将readBuf填充满NioSocketChannel后接下来轮询这个list去调用下面这个方法
pipeline.fireChannelRead(readBuf.get(i));
我们当前pipeline中除了head和tail还包含一个特殊的ChannelInboundHandler-ServerBootstrapAcceptor，我们暂且忽略在handler链前方的handler直接看ServerBootstrapAcceptor，而ServerBootstrapAcceptor我们在上篇文章已经分析过了，这里再回忆下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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);
    }
}

首先Acceptor接收到的信息强制转换为channel，接着执行childChannel初始化的三板斧，最后将获取到的channel注册到childGroup上，其实到这里这整个流程已经串起来了。 Ps这里的channel实际上是NioSocketChannel
（1）Server端调用bind，执行NioServerSocketChannel的实例化、初始化和注册（注册到BossGroup,这个时候触发了boss eventLoop的启动，开始无限循环执行那三件事情），在初始化过程中塞进了一个特殊的handler-ServerBootstrapAcceptor；接着执行bind方法和地址绑定到一起
（2）当有新连接到来的时候，会被eventLoop捕获到。捕获到的channel最开始还是原生的nio channel需要包装成netty中的channel，接着针对接收到的每个channel调用pipeline上的handler中对应的方法，这个时候执行到了ServerBootstrapAcceptor的channelRead方法，而在这个方法中将channel注册到了childGroup（注册到WorkerGroup,这个时候触发了worker eventLoop的启动，开始无限循环执行那三件事情）
Ps:接收到的channel以后就托管给childGroup了，对应事件的轮询也要由childGroup中的eventLoop来执行

不过为了『追求卓越』我们还是深入看下这里的register方法
我们追着代码走到了这里

1
2
3
4
5
//MultithreadEventLoopGroup.class
 @Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

先来看看这个next()

1
2
3
4
@Override
public EventExecutor next() {
    return chooser.next();
}

这里的chooser是一个EventExecutorChooser类，它的next方法从eventGroup中选择一个eventLoop，找到了eventLoop之后就是普通的channel注册过程了，和上篇文章中讲到的注册过程完全相同，最终都会执行
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
这段代码是经典的nio注册channel的代码，可见netty其实说白了就是对jdk nio的封装，使他更为好用罢了
Ps可见无论是通过channel向eventLoop注册or eventLoop注册channel最终都是执行上面那段code

极其重要！！！Worker的reactor线程在注册NioSocketChannel过程启动（之后就去无限循环它的那三件事情），以后该channel的读写事件就可以由worker的reactor线程来处理了

在register的过程中会执行很多用户定义的回调方法，比如invokeHandlerAddedIfNeeded()会调用pipeline上handler的handlerAdded方法，注意这个时候已经是childGroup中了，所以根据用户初始时候的输入

1
2
3
4
5
.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
    }
})

此时要执行ChannelInitializer的handlerAdded方法，而该方法会调用我们组装过程中定义的initChannel方法，这个方法调用完成后ChannelInitializer会将自己从pipeline中删除。

小结：
1.new server channel
2.init server channel(3 steps)
— 2.1 in the 3rd step add Acceptor to bossGroup pipeline as a handler
— Ps:Acceptor#channelRead init newChannel which just connects,and register it on childGroup
3.register server channel to bossGroup
— 3.1 boss reactor thread start(do 3 things)
— 3.2 fireManyMethods such as invokeHandlerAddedIfNeeded,fireChannelRegistered
4.bind address to server channel
5.new connect come
6.bossGroup catch new channel and package it as a Netty Channel
7.fire server channel pipeline
8.run into Acceptor#channelRead and register new channel to childGroup(in this process child reactor thread start)

参考链接：
https://www.jianshu.com/p/0242b1d4dd21