江苏龙网

  1. 主页 > 生活百科 > >

netty 服务端启动流程源码详解( 七 )

Netty

我们来重点关注下 read 方法 。
NioUnsafe 是一个接口，有两个子类：NioByteUnsafe 和 NioMessageUnsafe 。
NioServerSocketChannel 继承自 AbstractNioMessageChannel，使用的是 NioMessageUnsafe 类 。
read 方法实现如下：
@Overridepublic 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;            // 触发 channel read             pipeline.fireChannelRead(readBuf.get(i));        }        readBuf.clear();        allocHandle.readComplete();        // 触发 read complete        pipeline.fireChannelReadComplete();    } finally {        // 省略    }}doReadMessages 在 NioServerSocketChannel 类中实现如下：
@Overrideprotected 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) {        // 省略    }    return 0;}这里就是 jdk nio 中的接收到一个新的客户端请求的方法实现 。
读取完成之后，触发 fireChannelRead，如下：
@Overridepublic final ChannelPipeline fireChannelRead(Object msg) {    AbstractChannelHandlerContext.invokeChannelRead(head, msg);    return this;}如下：
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);    EventExecutor executor = next.executor();    if (executor.inEventLoop()) {        next.invokeChannelRead(m);    } else {        executor.execute(new Runnable() {            @Override            public void run() {                next.invokeChannelRead(m);            }        });    }}


推荐阅读


上一篇：Netty 实战：如何实现 HTTP 服务器？

下一篇：Netty 实战：如何实现文件服务器？