Netty源碼分析之一次請求是如何到達channelRead的?

時間 2019-11-07

標籤 netty 源碼分析一次請求如何到達 channelread 欄目 Netty 简体版

原文原文鏈接

如下分析只講NIOjava

使用java nio作網絡編程大體流程以下react

這個流程有哪些能夠優化的空間？git

java nio使用簡介
 java nio 啓動源碼分析github

Netty是對java網絡框架的包裝，它自己確定也會有相似的處理流程。一定在這個方面作了本身的優化處理編程

Netty 使用入門
 Netty Hello world源碼分析bash

得到Selector

使用Netty的時候都會用到對應的EventLoopGroup,它實際上就完成了Selector的初始化過程網絡

Netty自定義了SelectionKey的集合，作了層包裝，實際將Selector只有1個SelectorKey的集合換成了默認的兩個集合多線程

得到Channel

使用Netty時會執行channel的類型，而後在執行bind方法時，此處就會對channel實行初始化框架

構建的方式爲 class.newInstance(),以NioServerSocketChannel爲例，它執行的就是對應的無參構造函數。socket

public NioServerSocketChannel() {
 		//newSocket即返回java的ServerSocketChannel
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
 }
 public NioServerSocketChannel(ServerSocketChannel channel) {
 		//指定當前channel用來接收鏈接請求,並在父類中指定爲非阻塞
        super(null, channel, SelectionKey.OP_ACCEPT);
        //javaChannel()即這裏的參數channel
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
複製代碼

緊接着Netty開始channel的初始化，在NioServerSocketChannel的pipeline最後添加了一個ChannelInboundHandlerAdapter即ServerBootstrapAcceptor,它會執有 childGroup和childHandler，childHandler即用戶自定義的channelHandler,而childGroup則是處理請求所用的EventLoop,此時整個pipeline的結構爲

childGroup爲源碼中字段的命名，對應爲group中傳遞的worker線程池

channel的註冊與監聽端口地址關聯

註冊即創建channel和Selector的關係，值得注意的是，註冊使用的線程池爲group,對應用戶傳入的線程池即boss線程池，註冊和端口、地址關聯則順着Netty的啓動流程進行

至此能夠看到，java nio所須要的準備工做都已經準備好了，剩下的就是等待事件發生以及處理髮生的事件。與普通java nio的不一樣之處在於

Netty準備了兩個線程池，channel註冊、端口綁定監聽的只用到了其中同一個線程池

等待事件發生

NioEventLoop實現了Executor，意味着它接受其它地方提交任務給它執行,execute的大體結構以下

//判斷當前正在執行的線程是不是Netty本身的eventLoop中保存的線程
boolean inEventLoop = inEventLoop();
  if (inEventLoop) {
    //往隊列裏添加任務
  	addTask(task);
  } else {
  	//這裏即運行NioEventLoop自身的run方法
	startThread();
  	addTask(task);
  }
複製代碼

NioEventLoop啓動線程執行run方法，總體結構以下

for (;;) {
 if (hasTasks()) {
    selectNow();
   } else {
    select(oldWakenUp);
   }
  processSelectedKeys();
  runAllTasks();
}
複製代碼

run循環處理的流程以下

值得注意的是，這是單個線程在運行，並且非本線程的任務一律不處理

boss線程的啓動時機

在啓動的過程當中，有ServerBootstrap來串起整個流程，它的執行線程爲主線程，而註冊事件都是交由線程池本身來執行的，用程序表達來說，就是執行了eventLoop本身的execute,此時執行線程一定不是EventLoop本身的線程，從而boss中的線程啓動，在隊列任務中完成註冊

新鏈接請求的到來

當NioServerSocketChannel綁定了端口以後，NioServerSocketChannel對應的NioEventLoop會等待channel發生事件。整個處理流程以下

讀取消息的內容，發生在NioServerSocketChannel，對於這個新的鏈接事件，則包裝成一個客戶端的請求channel做爲後續處理

protected int doReadMessages(List<Object> buf) throws Exception {
 		//1:獲取請求的channel
        SocketChannel ch = javaChannel().accept();

        try {
            if (ch != null) {
            	//2:包裝成一個請求，Socket channel返回
                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;
    }
複製代碼

返回的NioSocketChannel則完成自身channel的初始化，註冊感興趣的事件

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ);
}
複製代碼

回想到boss中的下一環即ServerBootstrapAcceptor,而它讀取消息的處理則是添加用戶本身的handler,並繼續完成註冊事件

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);
            }
        }
複製代碼