当前位置：首页 > article >正文

Netty源码学习4——服务端是处理新连接的netty的reactor模式

article 2024/11/15 14:41:10

零丶引入

在前面的源码学习中，梳理了服务端的启动，以及NioEventLoop事件循环的工作流程，并了解了Netty处理网络io重要的Channel ，ChannelHandler，ChannelPipeline。

这一篇将学习服务端是如何构建新的连接。

一丶网络包接收流程

当客户端发送的网络数据帧通过网络传输到网卡时，网卡的DMA引擎将网卡接收缓冲区中的数据拷贝到DMA环形缓冲区，数据拷贝完成后网卡硬件触发硬中断，通知操作系统数据已到达。

随后网卡中断处理程序将DMA环形缓冲区的数据拷贝到sk_buffer，sk_buffer位于内核中，它提供了一个缓冲区，使得网卡中断程序可以将他接收到的数据暂存起来，避免数据丢失和切换。

随后发起软中断，网络协议栈会处理数据包，对数据包进行解析，路由，分发（根据目的端口号，分发给对应的应用程序，通过网络编程套接字，应用程序可以监听指定端口号，并接受网络协议栈的数据包）

当新的连接建立时，网络协议处理栈会将这个连接的套接字标记为可读，并生成一个accept事件，这个事件通知应用程序有新的连接需要处理

当已经建立的连接上有数据到达时，网络协议处理栈会将套接字标记为刻度，并生成一个read事件，这个事件通知应用程序有数据可供读取

当应用程序向已经建立的连接写入数据时，如果写缓冲区有足够的空间，写操作会立即完成，不会产生write事件。但如果写缓冲区已满，那么写操作将被暂停，当写缓冲区有足够的空间时，write事件将被触发，通知应用程序可以继续写入数据。

也就是说netty 服务端程序会监听不同的网络事件，并进行处理，这也是源码学习的切入点！

二丶服务端NioEventLoop处理网络IO事件

如上是NioEventLoop的运行机制，在《Netty源码学习2——NioEventLoop的执行》中我们进行了大致流程的学习，这一篇我么主要关注其run中处理网络IO事件的部分。

无论是否优化，最终都是拿到就绪的SelectionKey，循环处理每一个就绪的网络事件，如下便是处理的逻辑：

可以看到无论是accept事件还是read事件都是调用AbstractNioChannel的Unsafe#read方法

Unsafe是对netty对底层网络事件处理的封装，下面我们先看下AbstractNioChannel的类图，可以看到NioServerSocketChannel，和NioSocketChannel都使用继承了AbstractNioChannel，只是父类有所不同

那么NioServerSocketChannel和NioSocketChannel是什么时候Accept or read事件感兴趣的昵？

三丶NioServerSocketChannel设置对accept事件感兴趣

重点在ServerBootstrap#bind中，此方法会调用doBind0

doBind0会调用Channel#bind，然后处理ChannelPipeline#bind的执行，由于bind是出站事件，将从DefaultChannelPipeline的TailContext开始执行，然后调用到HeadContext#bind方法，最终会调用NioServerSocketChannel的unsafe#bind方法

如下是NioServerSocketChannel的unsafe#bind的内容：

主要完成两部分操作：

调用java原生ServerSocketChannel#bind方法，进行端口绑定，这样操作系统网络协议栈在分发网络数据的时候，才直到该分发到这个端口的ServerSocketChannel

向EventLoop中提交一个pipeline.fireChannelActive()的任务，将在pipeline上触发channelActive方法，HeadContext#channelActive将被调用到

这里将调用到Channel#read方法，最终会调用到HeadContext#read

四丶服务端处理Accept事件

前面我们说到，NioEventLoop处理accept事件和read事件都是调用unsafe#read方法，如下是NioServerSocketChannel#unsafe的read方法

  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 (continueReading(allocHandle));
                } catch (Throwable t) {
                    exception = t;
                }
				
                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    // 触发channelRead
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                readBuf.clear();
                allocHandle.readComplete();
                // 触发channelReadComplete
                pipeline.fireChannelReadComplete();

               // 省略
            } finally {
               // 省略
            }
        }

这里出现一个RecvByteBufAllocator.Handle，这里不需要过多关注，在NioServerSocketChannel建立连接的过程中，它负责控制是否还需要继续读取数据

ServerSocketChannel类提供了accept()方法，用于接受客户端的连接请求，返回一个SocketChannel代表了一个底层的TCP连接。

如上将jdk SocketChannel包装NioSocketChannel的时候会设置SocketChannel非阻塞并在属性readInterestOp记录感兴趣事件为read

包装生成的NioSocketChannel会放到List中，后续每一个就绪的连接会一次传播ChannelRead，并最终传播ChannelReadComplete

1.channeRead事件的传播

上面说到NioEventLoop读取NioServerSocketChannel上的accept事件，将每一个新连接封装为NioServerChannel后，将依次触发channelRead。

如下是ServerBootstrapAcceptor#channelRead方法，可以看到它会将读取生成的NioServerChannel注册到childGroup，这里的childGroup就是ServerBootstrap启动时候指定EventLoopGroup（主从reactor模式中的从reactor）

也就是说主reactor负责处理accept事件，从reactor负责处理read事件

2.channelReadComplete事件传播

大多数人看到 channelReadComplete 都会认为这是 Netty 读取了完整的数据，然而有时却不是这样。channelReadComplete 其实只是表明了本次从 Socket 读了数据，该方法通常可以用来进行一些收尾工作，例如发送响应数据或进行资源的释放等。channelReadComplete方法在每次读取数据完成后，即使没有更多的数据可读，也会被调用一次。

五丶netty对多种reactor模式的支持

这里其实可以看出netty对多种reactor模式（单线程，多线程，主从reactor）的支持

我们其实可以通过修改bossGroup，和workerGroup使netty使用不同的reactor模式

六丶将NioSocketChannel注册到从reactor

上面我们说到主reactor监听accept事件后传播channelRead事件，最终由ServerBootstrapAcceptor调用childGroup#register将包装生成的NioSocketChannel注册到从reactor（也就是workerGroup——EventLoopGroup）下面我们看看这个注册会发生什么

首先workerGroup这个EventLoopGroup会调用next方法选择出一个EventLoop执行register，然后

将NioSocketChannel中的jdk SockectChannel注册到Selector中，并将NioSocketChannel当作附件，这样selector#select到事件的时候，可以从附件中拿到网络事件对应的NioSocketChannel

触发handlerAdd

这一步触发ChannelHandler#handlerAdded

最终会调用到childHandler中指定的ChannelInitializer，它会将我们指定的ServerHandler(这里可以扩展我们的业务处理逻辑)加到NioSockectChannel的pipeline中

触发ChannelRegistered

触发channelActive

由于这是一个新连接，是第一次注册到EventLoop，因此会触发channelActive

这将调用到DefaultChannelPipeline的HeadContext#readIfIsAutoRead，最终就和我们第三节的【NioServerSocketChannel设置对accept事件感兴趣】差不多——HeadContext#readIfIsAutoRead会调用NioSockectChannel的read方法，最终调用到NioSockectChannel#unsafe的read方法——将注册对read事件感兴趣

七丶再看Netty的Reactor模式

笔者认为netty的reactor有以下几个要点

ServerBootstrap#bind方法

不仅仅会绑定端口，还会触发channelActive事件，从而使DefaultChannelPipeline中的HeadContext触发netty channel unsafe#beginRead,注册ServerSockectChannel对accept感兴趣

NioEventLoop处理新连接

这一步Netty 使用Selector进行IO多路复用，当accept事件产生的时候，调用NioServerSocketChannel#unsafe的read方法，这一步会将新连接封装NioSocketChannel，然后将对应连接的套接字注册到Selector上，然后传播channeRead事件