Netty源碼分析第三章: 客戶端接入流程
第四節: NioSocketChannel注冊到selector
我們回到最初的NioMessageUnsafe的read()方法:
public void read() { //必須是NioEventLoop方法調用的, 不能通過外部線程調用
assert eventLoop().inEventLoop(); //服務端channel的config
final ChannelConfig config = config(); //服務端channel的pipeline
final ChannelPipeline pipeline = pipeline(); //處理服務端接入的速率
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); //設置配置
allocHandle.reset(config); boolean closed = false; Throwable exception = null; try { try { do { //創建jdk底層的channel //readBuf用於臨時承載讀到鏈接
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; //傳遞事件, 將創建NioSokectChannel進行傳遞 //最終會調用ServerBootstrap的內部類ServerBootstrapAcceptor的channelRead()方法
pipeline.fireChannelRead(readBuf.get(i)); } readBuf.clear(); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); //代碼省略
} finally { //代碼省略
} }
在while循環結束之后, 將會通過一個for循環遍歷readBuf集合, 並將創建的NioSocketChannel傳入fireChannelRead()中, 傳播channel的讀取事件
有關pipeline的知識, 我們下一章會詳細剖析, 並會根據剖析后的內容回顧之前的有關pipeline的操作, 這里我們只需知道, 通過fireChannelRead()我們最終調用了ServerBootstrap的內部類ServerBootstrapAcceptor 中的channelRead()方法
跟到channelRead()方法中:
public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; //代碼省略
try { //work線程注冊channel
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); } }
其中參數的msg就是最初傳入fireChannelRead()方法的NioSocketChannel
所以這里可以通過 final Channel child = (Channel) msg 這種方式拿到NioSocketChannel
其中childGroup是我們最初初始化的work線程, 這里的register()方法跟boss線程一樣, 通過next()方法獲選擇一個線程進行注冊, 這里不再贅述
我們緊跟調用鏈, 跟到SingleThreadEventLoop的register()方法:
public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise"); promise.channel().unsafe().register(this, promise); return promise; }
這里的unsafe(), 根據我們之前的剖析, 是NioByteUnsafe, 這里的register最終會調用AbstractUnsafe的register()方法, 並NioSocketChannel
不知道同學們是否記得, 當初NioServerSocketChannel注冊的時候也走的這個方法
我們跟到register()這個方法中:
public final void register(EventLoop eventLoop, final ChannelPromise promise) { //省略驗證代碼 //所有的復制操作, 都交給eventLoop處理
AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop()) { //做實際主注冊
register0(promise); } else { try { eventLoop.execute(new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { //代碼省略
} } }
我們學習過NioEventLoop相關知識之后, 應該對這部分代碼不太陌生, 首先判斷是不是當前NioEventLoop線程, 如果是, 則直接進行注冊操作, 如果不是, 則封裝成task在當前NioEventLoop中執行
走到這里不難明白, 這里並不是當前NioEventLoop線程, 這是boss線程執行的, 所以這里會走到else, 如果是第一次的連接操作, work線程的NioEventLoop並沒有啟動, 所以這里也會啟動NioEventLoop, 並開始輪詢操作
跟到register0(promise)中看其是如何做實際操作的:
private void register0(ChannelPromise promise) { try { //省略代碼 //做實際的注冊
doRegister(); neverRegistered = false; registered = true; //觸發事件
pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); //觸發注冊成功事件
pipeline.fireChannelRegistered(); if (isActive()) { if (firstRegistration) { //傳播active事件(4)
pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { //省略代碼
} }
這段代碼我們同樣並不陌生, 因為NioServerSokectChannel中也走這一部分, 我們繼續關注doRegister()方法:
protected void doRegister() throws Exception { boolean selected = false; for (;;) { try { //jdk底層的注冊方法 //第一個參數為selector, 第二個參數表示不關心任何事件
selectionKey = javaChannel().register(eventLoop().selector, 0, this); return; } catch (CancelledKeyException e) { //省略代碼
} } }
這部分也是我們之前剖析過的jdk底層的注冊, 只是不同的是, 這里的javaChannel()是SocketChanel而不是ServerSocketChannel
同樣, 這里也是表示不關心任何事件, 只是在當前NioEventLoop綁定的selector上注冊
至此, NioSocketChannel完成注冊