Netty源碼分析（九）—IO事件處理流程

Netty的IO事件主要包括Read讀事件，Write寫事件和Connect連接配接事件；

個人首頁：tuzhenyu’s page

原文位址：Netty源碼分析（九）—IO事件處理流程

(0) 讀寫事件執行個體

服務端監聽8080端口，如果收到用戶端的時間查詢指令則傳回目前時間到用戶端

public class TimeServer {
    private void bind(int port){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try{
            ServerBootstrap b= new ServerBootstrap();
            b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new TimeServerHandler());
                    }
                });

            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        new TimeServer().bind();
    }
}

class TimeServerHandler extends ChannelInboundHandlerAdapter{

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf)msg;
        byte[] bytes = new byte[buf.readableBytes()];
        buf.readBytes(bytes);
        String body = new String(bytes,"UTF-8");
        System.out.println("the client says: "+ body);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)?new Date().toString():"BAD ORDER";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

}

用戶端在連接配接服務端成功後發送時間查詢指令，并接收服務端發來的消息

public class TimeClient {
    private void connect(String host,int port){
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new TimeClientHandler());
                        }
                    });

            ChannelFuture f = b.connect(host,port).sync();
            f.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }

    }

    public static void main(String[] args) {
        new TimeClient().connect("127.0.0.1",);
    }
}

class TimeClientHandler extends ChannelInboundHandlerAdapter{

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        byte[] bytes = "QUERY TIME ORDER".getBytes();
        ByteBuf buf = Unpooled.buffer(bytes.length);
        buf.writeBytes(bytes);
        ctx.writeAndFlush(buf);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf)msg;
        byte[] bytes = new byte[buf.readableBytes()];
        buf.readBytes(bytes);
        String body = new String(bytes,"UTF-8");
        System.out.println("the server says: "+ body);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

(1) IO事件處理入口

通道注冊到線程池後會開啟被綁定的線程監測被綁定的端口，線上程的run()方法中循環處理IO時間和任務隊列中的任務；

protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));

                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
            }

            cancelledKeys = ;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            if (ioRatio == ) {
                try {
                    processSelectedKeys();
                } finally {
                    runAllTasks();
                }
            } else {
                final long ioStartTime = System.nanoTime();
                try {
                    processSelectedKeys();
                } finally {
                    final long ioTime = System.nanoTime() - ioStartTime;
                    runAllTasks(ioTime * ( - ioRatio) / ioRatio);
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        try {
            if (isShuttingDown()) {
                closeAll();
                if (confirmShutdown()) {
                    return;
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
    }
}

當選擇器selector監測到有IO事件到來會通過processSelectedKeys()方法處理IO事件

private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized();
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

private void processSelectedKeysOptimized() {
    for (int i = ; i < selectedKeys.size; ++i) {
        final SelectionKey k = selectedKeys.keys[i];
        // null out entry in the array to allow to have it GC'ed once the Channel close
        // See https://github.com/netty/netty/issues/2363
        selectedKeys.keys[i] = null;

        final Object a = k.attachment();

        if (a instanceof AbstractNioChannel) {
            processSelectedKey(k, (AbstractNioChannel) a);
        } else {
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task);
        }

        if (needsToSelectAgain) {
            // null out entries in the array to allow to have it GC'ed once the Channel close
            // See https://github.com/netty/netty/issues/2363
            selectedKeys.reset(i + );

            selectAgain();
            i = -;
        }
    }
}

在時間處理時會判斷事件的類型，并進行相應的處理

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) {
            return;
        }
        if (eventLoop != this || eventLoop == null) {
            return;
        }
        unsafe.close(unsafe.voidPromise());
        return;
    }

    try {
        int readyOps = k.readyOps();
        if ((readyOps & SelectionKey.OP_CONNECT) != ) {
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }
        if ((readyOps & SelectionKey.OP_WRITE) != ) {
            ch.unsafe().forceFlush();
        }

        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) !=  || readyOps == ) {
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

(2) 讀事件的流程分析

處理讀事件，如果事件的類型是READ則調用read()方法處理讀事件

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

調用read()方法處理讀事件主要步驟包括：1.從記憶體池中配置設定記憶體擷取ByteBuf對象 2.從socket資料流中讀取資料到ByteBuf中 3.調用fireChannelRead()方法在pipeline管道中傳遞資料；

public final void read() {
        final ChannelConfig config = config();
        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);    //記憶體池中建立ByteBuf
                allocHandle.lastBytesRead(doReadBytes(byteBuf));
                if (allocHandle.lastBytesRead() <= ) {
                    // nothing was read. release the buffer.
                    byteBuf.release();
                    byteBuf = null;
                    close = allocHandle.lastBytesRead() < ;
                    break;
                }

                allocHandle.incMessagesRead();
                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對象

調用記憶體配置設定器ByteBufAllocator的allocate()方法配置設定記憶體擷取ByteBuf緩沖對象；

public ByteBuf allocate(ByteBufAllocator alloc) {
    return alloc.ioBuffer(guess());
}

在IO事件進行中ByteBuf緩沖區預設使用的是記憶體池中的直接記憶體；

public ByteBuf ioBuffer(int initialCapacity) {
    if (PlatformDependent.hasUnsafe()) {
        return directBuffer(initialCapacity);
    }
    return heapBuffer(initialCapacity);
}

public ByteBuf directBuffer(int initialCapacity) {
    return directBuffer(initialCapacity, DEFAULT_MAX_CAPACITY);
}

public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
    if (initialCapacity ==  && maxCapacity == ) {
        return emptyBuf;
    }
    validate(initialCapacity, maxCapacity);
    return newDirectBuffer(initialCapacity, maxCapacity);
}

從對象池中擷取ByteBuf緩沖對象，再從記憶體池中配置設定需要的記憶體

protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    PoolThreadCache cache = threadCache.get();
    PoolArena<ByteBuffer> directArena = cache.directArena;

    final ByteBuf buf;
    if (directArena != null) {
        buf = directArena.allocate(cache, initialCapacity, maxCapacity);
    } else {
        buf = PlatformDependent.hasUnsafe() ?
                UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
                new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
    }

    return toLeakAwareBuffer(buf);
}

從socket資料流中讀取資料到ByteBuf中

從socket資料流中讀取資料并寫入到緩沖區ByteBuf中

protected int doReadBytes(ByteBuf byteBuf) throws Exception {
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}

public int writeBytes(ScatteringByteChannel in, int length) throws IOException {
    ensureAccessible();
    ensureWritable(length);
    int writtenBytes = setBytes(writerIndex, in, length);
    if (writtenBytes > ) {
        writerIndex += writtenBytes;
    }
    return writtenBytes;
}

public int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
    checkIndex(index, length);
    ByteBuffer tmpBuf = internalNioBuffer();
    index = idx(index);
    tmpBuf.clear().position(index).limit(index + length);
    try {
        return in.read(tmpBuf);
    } catch (ClosedChannelException ignored) {
        return -;
    }
}

調用fireChannelRead()方法在pipeline管道中傳遞資料

調用fireChannelRead()方法将ByteBuf當參數傳遞進去，在pipeline中傳遞逐個經過定義的Handler處理

public 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處理IO事件的整體流程，主要步驟包括：
- 将通道注冊到線程池後會開啟通道綁定的線程，線上程中循環監測處理IO事件，同時也處理任務隊列中的任務；
- 監測到IO事件後調用processSelectedKeys()方法根據事件類型處理IO事件，如果是讀事件則調用read()方法處理；
- 在read()方法中先從記憶體池中配置設定擷取ByteBuf執行個體，然後從Socket資料流中讀取資料并寫入ByteBuf中，然後調用fireChannelRead()方法将ByteBuf在pipeline管道中傳遞，依次經過管道中定義的Handler處理；

Netty源碼分析（九）—IO事件處理流程Netty源碼分析（九）—IO事件處理流程

Netty源碼分析（九）—IO事件處理流程

(0) 讀寫事件執行個體

(1) IO事件處理入口

(2) 讀事件的流程分析

從記憶體池中配置設定記憶體擷取ByteBuf對象

從socket資料流中讀取資料到ByteBuf中

調用fireChannelRead()方法在pipeline管道中傳遞資料

總結

繼續閱讀

Netty-服務端channl的建立(源碼分析)服務端啟動的流程

Netty（七）源碼解析之 Reactor 模型、Netty的服務端啟動源碼分析1. Reactor 模型2. Netty 服務端啟動

netty源碼分析2-1-server啟動

Netty-NioEventLoop 源碼分析建立NioEventLoopNioEventLoop的啟動NioEventLoop執行邏輯

同樣都是程式員，為什麼别人的工資那麼高？

《Netty源碼學習一：pipleLine、FrameDecoder、Reactor多線程模型思想》Netty入門代碼例子Netty中的消息如何在管道pipeline中流轉如何了解Netty的編解碼器FrameDecoder從NIO Reactor多線程模型了解Netty Reactor模型

我教你一分鐘如何撬開阿裡爸爸的大門（想進阿裡的都看過來）1：面試前的準備2：面試中的遭遇3：面試後的感想

netty4學習筆記5-netty作為WebSocket伺服器

Netty 4 實踐 (1)初識netty4

Netty4之業務線程池的使用

[Android初級]android與netty4初體驗

netty4 學習入門1-netty作為http伺服器

【初學與研發之NETTY】netty4之檔案上傳

netty4 學習筆記3- netty的socket程式設計

史上最不詳細的Netty源碼分析-Netty啟動流程

netty學習之Channel接口