【对话写 Netty 代码的同学，你真的懂 Netty 了吗？四】client 端连接处理

拼搏现实的明天。 2022-10-17 00:49 19阅读 0赞

### 往期地址： ###

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（一）】Netty 初始化总览][Netty _ Netty _Netty]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（二）】主线流程 new NioEventLoopGroup(nThreads) 究竟做了什么][Netty _ Netty _ new NioEventLoopGroup_nThreads_]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（三上）】initAndRegister() 之 init() 怒了！这文章为啥写成了这样？？][Netty _ Netty _initAndRegister_ _ init_]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（三中）】initAndRegister() 之 init() 对不起！！我柜子动了…我不学了][Netty _ Netty _initAndRegister_ _ init_ 1]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？(三下)】server register 将何去何从？][Netty _ Netty _server register]

### 写在前面 ###

Netty 系列的第三部分，我们详细跟进了 NettyServer 端的初始化流程，从 init 到 register 再到 bind。

这里我们记住一个流程：

1.  循环监听的 select 在 init 时进行初始化，初始化方式是 new ChannelInitializer 并重写了 initChannel 方法，并将其先加入了 pipline 。
2.  在 register 流程中通过执行 pipeline.invokeHandlerAddedIfNeeded() 对 init 添加的 ChannelInitializer 进行了调用
3.  开始 select 的事件监听，并从 pipline 中移除由 init 时添加的 ChannelInitializer，同时 initChannel 逻辑中又将 ServerBootstrapAcceptor 加入了 pipline。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70]  
    在服务端启动后 pipline 是这个样子的：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 1]

### Client 端启动 ###

当 NettyServer 端启动后，select 进入了循环监听，这个时候就需要等待客户端的连接来触发 select 监听事件并进入对应的事件处理逻辑了。  
这里我们直接从客户端连接事件进来，触发 select 的事件 key 处理逻辑开始，如果看到这里有的蒙的同学建议回顾下之前的 init 第三部分中篇  
[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（三中）】initAndRegister() 之 init() 对不起！！我柜子动了…我不学了][Netty _ Netty _initAndRegister_ _ init_ 1]

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { 
            final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
            if (!k.isValid()) { 
                final EventLoop eventLoop;
                try { 
                    eventLoop = ch.eventLoop();
                } catch (Throwable ignored) { 
                    // If the channel implementation throws an exception because there is no event loop, we ignore this
                    // because we are only trying to determine if ch is registered to this event loop and thus has authority
                    // to close ch.
                    return;
                }
                // Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
                // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
                // still healthy and should not be closed.
                // See https://github.com/netty/netty/issues/5125
                if (eventLoop != this || eventLoop == null) { 
                    return;
                }
                // close the channel if the key is not valid anymore
                unsafe.close(unsafe.voidPromise());
                return;
            }
    
            try { 
                int readyOps = k.readyOps();
                // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
                // the NIO JDK channel implementation may throw a NotYetConnectedException.
                if ((readyOps & SelectionKey.OP_CONNECT) != 0) { 
                    // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                    // See https://github.com/netty/netty/issues/924
                    int ops = k.interestOps();
                    ops &= ~SelectionKey.OP_CONNECT;
                    k.interestOps(ops);
    
                    unsafe.finishConnect();
                }
    
                // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
                if ((readyOps & SelectionKey.OP_WRITE) != 0) { 
                    // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                    ch.unsafe().forceFlush();
                }
    
                // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
                // to a spin loop
                if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { 
                    unsafe.read();
                }
            } catch (CancelledKeyException ignored) { 
                unsafe.close(unsafe.voidPromise());
            }
        }

这里会走到连接事件

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { 
        unsafe.read();
    }

这里开启 NettyServer 后，运行 NettyClient 进行连接，通过 debug 可以看到监听到 readyOps 是 16 即 OP\_ACCEPT 事件，同时这里的 unsafe 是 NioMessageUnsafe  
![在这里插入图片描述][20210603154616959.png]  
进入 read() 方法

private final class NioMessageUnsafe extends AbstractNioUnsafe { 
    
            private final List<Object> readBuf = new ArrayList<Object>();
    
            @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));
                    }
                    readBuf.clear();
                    allocHandle.readComplete();
                    pipeline.fireChannelReadComplete();
    
                    if (exception != null) { 
                        closed = closeOnReadError(exception);
    
                        pipeline.fireExceptionCaught(exception);
                    }
    
                    if (closed) { 
                        inputShutdown = true;
                        if (isOpen()) { 
                            close(voidPromise());
                        }
                    }
                } finally { 
                    // Check if there is a readPending which was not processed yet.
                    // This could be for two reasons:
                    // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
                    // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
                    //
                    // See https://github.com/netty/netty/issues/2254
                    if (!readPending && !config.isAutoRead()) { 
                        removeReadOp();
                    }
                }
            }
        }

核心方法：

int localRead = doReadMessages(readBuf);
    pipeline.fireChannelRead(readBuf.get(i));
    pipeline.fireChannelReadComplete();

首先看下 doReadMessages 方法这里传入了一个 list

private final List<Object> readBuf = new ArrayList<Object>();
    int localRead = doReadMessages(readBuf);

这里是 NioServerSocketChannel 的实现

@Override
        protected int doReadMessages(List<Object> buf) throws Exception { 
            SocketChannel ch = SocketUtils.accept(javaChannel());
    
            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;
        }

最终将 accept 的 SocketChannel 封装成 NioSocketChannel 添加到了列表 readBuf 中

SocketChannel ch = SocketUtils.accept(javaChannel());
    buf.add(new NioSocketChannel(this, ch));

这里就初始化了一个 NioSocketChannel ，其逻辑类似于，NioServerSocketChannel 的初始化

public NioSocketChannel(Channel parent, SocketChannel socket) { 
            super(parent, socket);
            config = new NioSocketChannelConfig(this, socket.socket());
        }
        protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) { 
            super(parent, ch, SelectionKey.OP_READ);
        }
        protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { 
            super(parent);
            this.ch = ch;
            this.readInterestOp = readInterestOp;
            try { 
                ch.configureBlocking(false);
            } catch (IOException e) { 
                try { 
                    ch.close();
                } catch (IOException e2) { 
                    if (logger.isWarnEnabled()) { 
                        logger.warn(
                                "Failed to close a partially initialized socket.", e2);
                    }
                }
    
                throw new ChannelException("Failed to enter non-blocking mode.", e);
            }
        }
        protected AbstractChannel(Channel parent) { 
            this.parent = parent;
            id = newId();
            unsafe = newUnsafe();
            pipeline = newChannelPipeline();
        }

`SelectionKey.OP_READ` 这里注册了 OP\_READ

这时候会调用 NioServerSocket 的 pipline 的 fireChannelRead 方法，这里会调用链路中每个 handler 的 channelRead 方法，那么这里在之前的初始化逻辑中可以知道该 pipline 中只有一个即 ServerBootstrapAcceptor

pipeline.fireChannelRead(readBuf.get(i));

调用 ServerBootstrapAcceptor 的 channelRead 方法

@Override
            @SuppressWarnings("unchecked")
            public void channelRead(ChannelHandlerContext ctx, Object msg) { 
                final Channel child = (Channel) msg;
    
                child.pipeline().addLast(childHandler);
    
                setChannelOptions(child, childOptions, logger);
    
                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);
                }
            }

1.  获取新初始化的 NioSocketChannel
2.  在 NioSocketChannel 的 pipline 中添加 childHandler （此 childHandler 即我们初始化时自己定制的 Handler）
3.  childGroup 即 workerGroup 调用 register 注册新初始化的 NioSocketChannel

final Channel child = (Channel) msg; // 
    child.pipeline().addLast(childHandler);
    childGroup.register(child).addListener（……）

这里的 register 和之前 Server 的注册逻辑是几乎一模一样的，走的都是同一个类的同一个方法  
SingleThreadEventLoop 的 register  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 2]  
不过这个时候走的是提交任务的逻辑，这里 execute 会进行 select 的循环监听。

到这里客户端的连接注册已经完成了。

当有读事件发生的时候同样会走到这块的逻辑，但这个循环 select 是上一步初始化的 workerGroup 的 NioEventLoop 循环。而此时的 unsafe 是 NioByteUnsafe

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { 
        unsafe.read();
    }

> 这里如何知道 unsafe 实现是 NioByteUnsafe 有两种方法
> 
> 1.  打断点 debug 查看
> 2.  跟踪静态代码查看初始化逻辑

第一点就不说了，这里稍微说下第二点，这里是 NioSocketChannel ，所以我们直接定位到该类的初始化可以看到 初始化的是 `NioSocketChannelUnsafe extends NioByteUnsafe`

@Override
        protected AbstractNioUnsafe newUnsafe() { 
            return new NioSocketChannelUnsafe();
        }
        private final class NioSocketChannelUnsafe extends NioByteUnsafe { 
            @Override
            protected Executor prepareToClose() { 
                try { 
                    if (javaChannel().isOpen() && config().getSoLinger() > 0) { 
                        // We need to cancel this key of the channel so we may not end up in a eventloop spin
                        // because we try to read or write until the actual close happens which may be later due
                        // SO_LINGER handling.
                        // See https://github.com/netty/netty/issues/4449
                        doDeregister();
                        return GlobalEventExecutor.INSTANCE;
                    }
                } catch (Throwable ignore) { 
                    // Ignore the error as the underlying channel may be closed in the meantime and so
                    // getSoLinger() may produce an exception. In this case we just return null.
                    // See https://github.com/netty/netty/issues/4449
                }
                return null;
            }
        }

进入 NioByteUnsafe 的 read 方法，看看当 NioSocketChannel 监听到 read 事件后做了什么

@Override
            public final void read() { 
                final ChannelConfig config = config();
                if (shouldBreakReadReady(config)) { 
                    clearReadPending();
                    return;
                }
                final ChannelPipeline pipeline = pipeline();
                final ByteBufAllocator allocator = config.getAllocator();
                final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
                allocHandle.reset(config);
    
                ByteBuf byteBuf = null;
                boolean close = false;
                try { 
                    do { 
                        byteBuf = allocHandle.allocate(allocator);
                        allocHandle.lastBytesRead(doReadBytes(byteBuf));
                        if (allocHandle.lastBytesRead() <= 0) { 
                            // nothing was read. release the buffer.
                            byteBuf.release();
                            byteBuf = null;
                            close = allocHandle.lastBytesRead() < 0;
                            if (close) { 
                                // There is nothing left to read as we received an EOF.
                                readPending = false;
                            }
                            break;
                        }
    
                        allocHandle.incMessagesRead(1);
                        readPending = false;
                        pipeline.fireChannelRead(byteBuf);
                        byteBuf = null;
                    } while (allocHandle.continueReading());
    
                    allocHandle.readComplete();
                    pipeline.fireChannelReadComplete();
    
                    if (close) { 
                        closeOnRead(pipeline);
                    }
                } catch (Throwable t) { 
                    handleReadException(pipeline, byteBuf, t, close, allocHandle);
                } finally { 
                    // Check if there is a readPending which was not processed yet.
                    // This could be for two reasons:
                    // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
                    // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
                    //
                    // See https://github.com/netty/netty/issues/2254
                    if (!readPending && !config.isAutoRead()) { 
                        removeReadOp();
                    }
                }
            }
        }

这里我们关注核心代码

ByteBuf byteBuf = null;
    byteBuf = allocHandle.allocate(allocator);
    allocHandle.lastBytesRead(doReadBytes(byteBuf));
    pipeline.fireChannelRead(byteBuf);
    pipeline.fireChannelReadComplete();

相信看过之前文章的同学对这些方法都很熟悉了，这里我们直接说逻辑

1.  `allocHandle.allocate(allocator)` 为 ByteBuf 分配空间 (这里涉及到 零拷贝 我们放到后面说)
2.  `allocHandle.lastBytesRead(doReadBytes(byteBuf))` 从管道中读取数据到 byteBuf 中
3.  `pipeline.fireChannelRead(byteBuf);` 调用 pipline 中所有 handler 的 channelRead 方法
4.  `pipeline.fireChannelReadComplete();` 调用 pipline 中所有 handler 的 channelReadComplete 方法

到这里 client 端的连接和读取事件的处理就完成了。

### 总结 ###

最后稍微总结一下

> 1 当新的客户端进行连接注册时 ，会被 bossGroup 监听到 SelectionKey.OP\_ACCEPT unsafe.read() 执行注册逻辑 ，此时的 unsafe 是 NioServerSocketChannel 初始化时的 NioMessageUnsafe  
> 2 注册会初始化 NioSocketChannel 将所有我们定制的 Handler 加入到 NioSocketChannel 的 pipline 中，并开启 select 进行监听 OP\_READ 事件  
> 3 客户端有数据传输，此时会触发 NioSocketChannel select 的监听 unsafe.read() 逻辑 此时的 unsafe 是 NioByteUnsafe ，此时会读取通道中的数据并依次调用 pipline 中我们定制的 handler 的 channelRead 方法 以及之后调用所有的 channelReadComplete 方法

到这里整体的流程就梳理完了，从服务端的启动，到客户端的连接和读取，后面我们聊聊一些细节和常问的问题 比如 零拷贝 等

感谢你能看到这里，希望本文对你有所帮助

**我是 [dying 搁浅][dying] ，我始终期待与你的相遇。无论你是否期待，潮涨潮落，我仅且就在这里…**

**欢迎添加个人微信 dyinggq 一起交流学习~~**

我们下期再见~  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 3]

[Netty _ Netty _Netty]: https://blog.csdn.net/w903328615/article/details/114899473
[Netty _ Netty _ new NioEventLoopGroup_nThreads_]: https://blog.csdn.net/w903328615/article/details/115680055
[Netty _ Netty _initAndRegister_ _ init_]: https://blog.csdn.net/w903328615/article/details/115982104
[Netty _ Netty _initAndRegister_ _ init_ 1]: https://blog.csdn.net/w903328615/article/details/116499703
[Netty _ Netty _server register]: https://blog.csdn.net/w903328615/article/details/116948452
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70]: /images/20221014/1f008826f92542909d0f940434a860c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 1]: /images/20221014/40cc879111024302a85d2a2ad92c75ac.png
[20210603154616959.png]: /images/20221014/ee067dc694b5485b92f86a68f2916d99.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 2]: /images/20221014/95189b2b99e94a9c95eb388dbfd3cc0e.png
[dying]: https://blog.csdn.net/w903328615
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 3]: /images/20221014/e79124ba148947ada7817b776d118f78.png