第五章 Netty服務器啟動過程 · Netty源碼筆記

服務器的啟動過程大量使用了EventLoop和Future/Promise，在閱讀源碼之前，建議首先要對Netty的這兩種機制進行了解。由于Netty更多是在服務器端使用，因此以服務器的啟動過程為例進行學習。 ## 5.1 階段：配置config 配置階段的工作很簡單，主要就是初始化啟動類，設置相關參數。 Bootstrap啟動類主要功能是初始化啟動器，為啟動器設置相關屬性。我們先來看一下Bootstrap的類結構，啟動類有一個AbstractBootstrap基類，有兩個實現類Bootstrap和ServerBootstrap，分別用于客戶端和服務器的啟動。 ### AbstractBootstrap **屬性** ``` EventLoopGroup group; //線程組,對于ServerBootstrap來說，group為ServerSocketChannel服務 ChannelFactory<? extends C> channelFactory; //用于獲取channel的工廠類 SocketAddress localAddress;//綁定的地址 Map<ChannelOption<?>, Object> options;//channel可設置的選項，包含java-channel和netty-channel Map<AttributeKey<?>, Object> attrs;//channel屬性，便于保存用戶自定義數據 ChannelHandler handler;//Channel處理器 ``` **方法** ``` group() 設置線程組 channelFactory()及channel() 設置channel工廠和channel類型 localAddress() 設置地址 option() 添加channel選項 attr() 添加屬性 handler() 設置channelHander ``` 上面這些方法主要用于設置啟動器的相關參數，除此之外，還有一些啟動時調用的方法 ``` register() 內部調用initAndRegister() 用來初始化channel并注冊到線程組 bind() 首先會調用initAndRegister()，之后綁定IP地址，使用Promise保證先initAndRegister()在bind() initAndRegister()，主要是創建netty的channel，設置options和attrs，注冊到線程組 ``` ### ServerBootstrap ServerBootstrap在AbstractBootstrap的基礎上添加了如下屬性，用來設置子Channel，也就是客戶端連接后創建的Channel的屬性。另外，還實現了抽象類中定義的init()方法。 ``` Map<ChannelOption<?>, Object> childOptions Map<AttributeKey<?>, Object> childAttrs EventLoopGroup childGroup; ChannelHandler childHandler; ``` ## 5.2 階段：初始化init 初始化init階段的主要功能是：創建并初始化服務器的Netty-Channel；分為兩個步驟：創建和初始化。 **創建NettyChannel** * 使用SelectorProvider打開java通道 * 為Channel分配全局唯一的ChannelID * 創建NioMessageUnsafe，用于netty底層的讀寫操作 * 創建ChannelPipeline，默認的是DefaultChannelPipeline 下面是初始init階段的主要代碼： ``` Channel channel = null; try { channel = channelFactory.newChannel();// 創建NettyChannel init(channel);//初始化NettyChannel } catch (Throwable t) { if (channel != null) { // channel can be null if newChannel crashed (eg SocketException("too many open files")) channel.unsafe().closeForcibly(); } // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } ``` channelFactory用于獲取Channel實例，啟動時，channelFactory在調用channel(NioServerSocketChannel.class)設置channel類型時創建,由于我們使用的是設置class的方法，會使用ReflectiveChannelFactory作為工廠類，其會直接調用class的newInstance獲取Channel實例。Netty中，服務器端的Channel為NioServerSocketChannel，客戶端為NioSocketChannel。 Channel的創建過程如下： 1. 打開java通道：NioServerSocketChannel創建時，首先使用SelectorProvider的openServerSocketChannel打開服務器套接字通道。SelectorProvider是Java的NIO提供的抽象類，是選擇器和可選擇通道的服務提供者。具體的實現類有SelectorProviderImpl，EPollSelectorProvide，PollSelectorProvider。選擇器的主要工作是根據操作系統類型和版本選擇合適的Provider：如果LInux內核版本>=2.6則，具體的SelectorProvider為EPollSelectorProvider，否則為默認的PollSelectorProvider。至此，底層的Java ServerSocketChannel創建完畢。 ``` public NioServerSocketChannel() { this(newSocket(DEFAULT_SELECTOR_PROVIDER)); } private static ServerSocketChannel newSocket(SelectorProvider provider) { return provider.openServerSocketChannel(); } ``` 2. Java ServerSocketChannel創建完畢后，會進入netty-Channel的構造方法,首先初始化ChannelId，ChannelId是一個全局唯一的值； 3. 之后，創建NioMessageUnsafe實例，該類為Channel提供了用于完成網絡通訊相關的底層操作，如connect(),read(),register(),bind(),close()等； 4. 為Channel創建DefaultChannelPipeline，初始化雙向鏈表； 5. 講java-channel設置為非阻塞，將關注的操作設置為SelectionKey.OP_ACCEPT（服務器） ``` protected DefaultChannelPipeline(Channel channel) { this.channel = ObjectUtil.checkNotNull(channel, "channel"); succeededFuture = new SucceededChannelFuture(channel, null); voidPromise = new VoidChannelPromise(channel, true); // 初始化雙向鏈表 tail = new TailContext(this); // 創建head head = new HeadContext(this); // 創建tail head.next = tail; tail.prev = head; } ``` **初始化NettyChannel** 創建NettyChannel后，下一步需要進行初始化，由于服務器端和客戶端的Channel不一樣，因此init方法被分別實現到了ServerBootstrap和Bootstrap中，我們主要分析服務器的init。服務器的init分為幾個步驟： * 將啟動器設置的選項和屬性設置到NettyChannel上面 * 向Pipeline添加初始化Handler，供注冊后使用具體實現在ServerBootstrap類的init方法中，程序比較簡單。每個NettyChannel對象保護一個ChannelConfig類保存相關配置，還有Map<AttributeKey<?>, Object> attrs用來保存自定義屬性。至于初始化Handler，我們先記住，在bind中會說明其作用。在addLast時，由于還未注冊，因此會加入到Pipeline的一個等待鏈表中，待注冊后執行。 ``` if (!registered) { newCtx.setAddPending(); callHandlerCallbackLater(newCtx, true); return this; } ``` 總結一下，這個階段的代碼我們可以看出，channel內部包含幾個重要對象： ``` ChannelID 全局唯一ID ChannelConfig 保存配置 ChannelPipeline 通道的流水線 Unsafe Netty底層封裝的網絡I/O操作 ``` ## 5.3 階段：注冊register 這個階段的主要工作是將創建并初始化后的NettyChannel注冊到selector上面。具體過程： * 將打開NettyChannel注冊到線程池組的selector上； * 觸發Pipeline上面ChannelHandler的channelRegistered， ``` // AbstractBootstrap類 initAndRegister() ChannelFuture regFuture = config().group().register(channel); ``` 上面的程序會使用傳入的線程池組的register(channel);注冊NettyChannel，具體方法定義在SingleThreadEventLoop中,其會使用NettyChannel的unsafe的register方法，該方法首先會判斷當前線程是否是指定線程池正在運行的線程，如果不是提交到要注冊的線程池中執行。執行時調用下面的程序。 ``` // AbstractUnsafe，刪去了部分校驗代碼 private void register0(ChannelPromise promise) { try { boolean firstRegistration = neverRegistered;// 是否為首次注冊 doRegister(); // 1. 注冊 neverRegistered = false; registered = true; pipeline.invokeHandlerAddedIfNeeded();// 2. 將注冊之前加入的handler加入進來 safeSetSuccess(promise); // 注冊成功，通知promise pipeline.fireChannelRegistered();// 4. Pipeline通知觸發注冊成功 if (isActive()) { // 是否已經綁定因為register和bind階段是異步的 if (firstRegistration) { pipeline.fireChannelActive(); // 5.首次注冊，通知 } else if (config().isAutoRead()) {// Channel會deregister后重新注冊到線程組時，且配置了AutoRead beginRead(); } } } catch (Throwable t) { closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } } ``` 1. 注冊：將NettyChannel內部的javaChannel注冊到線程池的selector上面，由線程池不斷執行select()查詢準備就緒的文件描述符。具體實現在AbstractNioChannel中的doRegister() 2. invokeHandlerAddedIfNeeded: 注冊成功后，找到初始化階段通過pipeline.addLast()加入的ChannelInitializer，執行其ChannelInitializer的initChannel方法，之后將其刪除（在ChannelInitializer的initChannel方法中）；初始化NettyChannel階段，我們addLast了一個初始化Handler，現在來看看其作用 ``` // init初始化階段添加了一個ChannelInitializer p.addLast(new ChannelInitializer<Channel>() { @Override public void initChannel(Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler();// 獲取config時設置的handler if (handler != null) { pipeline.addLast(handler); // 將其添加到鏈表尾部 } // 加入一個ServerBootstrapAcceptor處理器，用于處理Accept ch.eventLoop().execute(new Runnable() { @Override public void run() { pipeline.addLast(new ServerBootstrapAcceptor( currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); ``` 從上面的程序中可以看到，由于初始化時，還未將javaChannel注冊到線程池的selector上，此時還無法設置Channel將Accept注冊到選擇器上，因此先加入了一個ChannelInitializer，等待register后向Pipeline加入ServerBootstrapAcceptor。此時，NettyChannel的Pipeline的鏈表結構為: ``` Head <--> InitialHandler <--> ServerBootstrapAcceptor <--> Tail ``` 在initChannel執行的最后會將InitialHandler從Pipeline移除，此時NioServerSocketChannel的鏈表結構為 ``` Head <--> ServerBootstrapAcceptor <--> Tail ``` 3. fireChannelRegistered,沿著pipeline的head到tail，調用ChannelHandler的channelRegistered方法， ``` public final ChannelPipeline fireChannelRegistered() { AbstractChannelHandlerContext.invokeChannelRegistered(head); return this; } private void invokeChannelRegistered() { if (invokeHandler()) { // 狀態是否正確 try { ((ChannelInboundHandler) handler()).channelRegistered(this); // 觸發 } catch (Throwable t) { notifyHandlerException(t); } } else { fireChannelRegistered();// 狀態不正確，通知下一個Handler } } ``` 4. fireChannelActive 由于注冊階段和綁定bind階段都是異步的，如果此時注冊完成時bind階段已經綁定了本地端口，會沿著pipeline的head到tail，調用各個Handler的channelActive方法 ## 5.4 階段：綁定bind 本階段的主要內容是：將NettyChannel內部的java的ServerSocketChannel綁定到本地的端口上面，結束后使用fireChannelActive通知Pipeline里的ChannelHandle，執行其channelActive方法。 bind的入口為AbstractBootstrap的doBind0()，內部會調用pipeline中的bind方法，邏輯為從tail出發，調用outbound的ChannelHandler的bind方法，從上面我們可以看到當前的鏈表如下： ``` Head[I/O] <--> ServerBootstrapAcceptor[IN] <--> Tail[IN] ``` 只有Head可以用來處理Outbound，Head的bind方法調用了channel創建過程中生成的unsafe對象NioMessageUnsafe的實例，該實例的bind方法首先java的channel bind本地地址，然后觸發fireChannelActive。 ``` public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { boolean wasActive = isActive(); try { doBind(localAddress); } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { invokeLater(new Runnable() { @Override public void run() { pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); } ``` 至此，Netty的服務器段已經啟動，Channel和ChannelPipeline已經建立。EventLoop也在不斷的select()查找準備好的I/O。