摘要:而這個正是它是的內部類,同時繼承自。獲取最近的并依次執行其方法進入頭部,并且最終更改了向注冊了讀事件參考文章總結如何接入新連接基本流程如上所述,如果有誤,還望各位指正。
歡迎關注公眾號:【愛編程】
如果有需要后臺回復2019贈送1T的學習資料哦!!
前文再續,書接上一回【NioEventLoop】。
在研究NioEventLoop執行過程的時候,檢測IO事件(包括新連接),處理IO事件,執行所有任務三個過程。其中檢測IO事件中通過持有的selector去輪詢事件,檢測出新連接。這里復用同一段代碼。
在開始分析前,先了解一下Channel的設計
頂層Channel接口定義了socket事件如讀、寫、連接、綁定等事件,并使用AbstractChannel作為骨架實現了這些方法。查看器成員變量,發現大多數通用的組件,都被定義在這里
第二層AbstractNioChannel定義了以NIO,即Selector的方式進行讀寫事件的監聽。其成員變量保存了selector相關的一些屬性。
第三層內容比較多,定義了服務端channel(左邊繼承了AbstractNioMessageChannel的NioServerSocketChannel)以及客戶端channel(右邊繼承了AbstractNioByteChannel的NioSocketChannel)。
如何接入新連接?本文開始探索一下Netty是如何接入新連接?主要分為四個部分
1.檢測新連接1.檢測新連接
2.創建NioSocketChannel
3.分配線程和注冊Selector
4.向Selector注冊讀事件
Netty服務端在啟動的時候會綁定一個bossGroup,即NioEventLoop,在bind()綁定端口的時候注冊accept(新連接接入)事件。掃描到該事件后,便處理。因此入口從:NioEventLoop#processSelectedKeys()開始。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe(); //省略代碼 // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead // to a spin loop //如果當前NioEventLoop是workGroup 則可能是OP_READ,bossGroup是OP_ACCEPT if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { //新連接接入以及讀事件處理入口 unsafe.read(); } }
關鍵的新連接接入以及讀事件處理入口unsafe.read();
a).這里的unsafe是在Channel創建過程的時候,調用了父類AbstractChannel#AbstractChannel()的構造方法,和pipeline一起初始化的。
protected AbstractChannel(Channel parent) { this.parent = parent; id = newId(); unsafe = newUnsafe(); pipeline = newChannelPipeline(); }
服務端:
unsafe 為NioServerSockeChannel的父類AbstractNioMessageChannel#newUnsafe()創建,可以看到對應的是AbstractNioMessageChannel的內部類NioMessageUnsafe;
客戶端:
unsafe為NioSocketChannel的的父類AbstractNioUnsafe#newUnsafe()創建的話,它對應的是AbstractNioByteChannel的內部類NioByteUnsafe
b).unsafe.read()
NioMessageUnsafe.read()中主要的操作如下:
1.循環調用jdk底層的代碼創建channel,并用netty的NioSocketChannel包裝起來,代表新連接成功接入一個通道。
2.將所有獲取到的channel存儲到一個容器當中,檢測接入的連接數,默認是一次接16個連接
3.遍歷容器中的channel,依次調用方法fireChannelRead,4.fireChannelReadComplete,fireExceptionCaught來觸發對應的傳播事件。
private final class NioMessageUnsafe extends AbstractNioUnsafe { //臨時存儲讀到的連接 private final List
而這一段關鍵代碼邏輯中 int localRead = doReadMessages(readBuf);它創建jdk底層channel并且用NioSocketChannel包裝起來,將該channel添加到傳入的容器保存起來,同時返回一個計數。
protected int doReadMessages(List2.創建NioSocketChannelbuf) throws Exception { SocketChannel ch = SocketUtils.accept(javaChannel()); try { if (ch != null) { //將jdk底層的channel封裝到netty的channel,并存儲到傳入的容器當中 //this為服務端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; }
通過檢測IO事件輪詢新連接,當前成功檢測到連接接入事件之后,會調用NioServerSocketChannel#doReadMessages()方法,進行創建NioSocketChannel,即客戶端channel的過程。
下面就來了解一下NioSocketChannel的主要工作:
.查看原代碼做了兩件事,調用父類構造方法,實例化一個NioSocketChannelConfig。
public NioSocketChannel(Channel parent, SocketChannel socket) { super(parent, socket); //實例化一個NioSocketChannelConfig config = new NioSocketChannelConfig(this, socket.socket()); }
1)、查看NioSocketChannel父類構造方法,主要是保存客戶端注冊的讀事件、channel為成員變量,以及設置阻塞模式為非阻塞。
public NioSocketChannel(Channel parent, SocketChannel socket) { super(parent, socket); //實例化一個NioSocketChannelConfig config = new NioSocketChannelConfig(this, socket.socket()); } protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) { //傳入感興趣的讀事件:客戶端channel的讀事件 super(parent, ch, SelectionKey.OP_READ); } protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); //保存客戶端channel為成員變量 this.ch = ch; //保存感興趣的讀事件為成員變量 this.readInterestOp = readInterestOp; try { //配置阻塞模式為非阻塞 ch.configureBlocking(false); } catch (IOException e) { try { ch.close(); } catch (IOException e2) { if (logger.isWarnEnabled()) { logger.warn( "Failed to close a partially initialized socket.", e2); } } throw new ChannelException("Failed to enter non-blocking mode.", e); } }
最后調用父類的構造方法,是設置該客戶端channel對應的服務端channel,以及channel的id和兩大組件unsafe和pipeline
protected AbstractChannel(Channel parent) { //parent為創建次客戶端channel的服務端channel(服務端啟動過程中通過反射創建的) this.parent = parent; id = newId(); unsafe = newUnsafe(); pipeline = newChannelPipeline(); }
2)、再看NioSocketChannelConfig實例化。主要是保存了javaSocket,并且通過setTcpNoDelay(true);禁止了tcp的Nagle算法,目的是為了盡量讓小的數據包整合成大的發送出去,降低延時.
private NioSocketChannelConfig(NioSocketChannel channel, Socket javaSocket) { super(channel, javaSocket); calculateMaxBytesPerGatheringWrite(); } public DefaultSocketChannelConfig(SocketChannel channel, Socket javaSocket) { super(channel); if (javaSocket == null) { throw new NullPointerException("javaSocket"); } //保存socket this.javaSocket = javaSocket; // Enable TCP_NODELAY by default if possible. if (PlatformDependent.canEnableTcpNoDelayByDefault()) { try { //禁止Nagle算法,目的是為了讓小的數據包盡量集合成大的數據包發送出去 setTcpNoDelay(true); } catch (Exception e) { // Ignore. } } }3.分配線程和注冊Selector
服務端啟動初始化的時候ServerBootstrap#init(),主要做了一些參數的配置。其中對于childGroup,childOptions,childAttrs,childHandler等參數被進行了多帶帶配置。作為參數和ServerBootstrapAcceptor一起,被當作一個特殊的handle,封裝到pipeline中。ServerBootstrapAcceptor中的eventLoop為workGroup。
public class ServerBootstrap extends AbstractBootstrap{ //省略了很多代碼............. @Override void init(Channel channel) throws Exception { //配置AbstractBootstrap.option final Map , Object> options = options0(); synchronized (options) { setChannelOptions(channel, options, logger); } //配置AbstractBootstrap.attr final Map , Object> attrs = attrs0(); synchronized (attrs) { for (Entry , Object> e: attrs.entrySet()) { @SuppressWarnings("unchecked") AttributeKey key = (AttributeKey ) e.getKey(); channel.attr(key).set(e.getValue()); } } //配置pipeline ChannelPipeline p = channel.pipeline(); //獲取ServerBootstrapAcceptor配置參數 final EventLoopGroup currentChildGroup = childGroup; final ChannelHandler currentChildHandler = childHandler; final Entry , Object>[] currentChildOptions; final Entry , Object>[] currentChildAttrs; synchronized (childOptions) { currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0)); } synchronized (childAttrs) { currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0)); } p.addLast(new ChannelInitializer () { @Override public void initChannel(final Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); //配置AbstractBootstrap.handler ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { //配置ServerBootstrapAcceptor,作為Handle緊跟HeadContext pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); } //省略了很多代碼............. }
可見,整個服務端pipeline的結構如下圖所示。bossGroup控制IO事件的檢測與處理,整個bossGroup對應的pipeline只包括頭(HeadContext)尾(TailContext)以及中部的ServerBootstrap.ServerBootstrapAcceptor。
當新連接接入的時候AbstractNioMessageChannel.NioMessageUnsafe#read()方法被調用,最終調用fireChannelRead(),方法來觸發下一個Handler的channelRead方法。而這個Handler正是ServerBootstrapAcceptor
它是ServerBootstrap的內部類,同時繼承自ChannelInboundHandlerAdapter。也是一個ChannelInboundHandler。其中channelRead主要做了以下幾件事。
1.為客戶端channel的pipeline添加childHandler
2.設置客戶端TCP相關屬性childOptions和自定義屬性childAttrs
3.workGroup選擇NioEventLoop并注冊Selector
1)、為客戶端channel的pipeline添加childHandler
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter { private final EventLoopGroup childGroup; private final ChannelHandler childHandler; private final Entry, Object>[] childOptions; private final Entry , Object>[] childAttrs; private final Runnable enableAutoReadTask; ServerBootstrapAcceptor( final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler, Entry , Object>[] childOptions, Entry , Object>[] childAttrs) { this.childGroup = childGroup; this.childHandler = childHandler; this.childOptions = childOptions; this.childAttrs = childAttrs; //省略了一些代碼。。。。。 @Override @SuppressWarnings("unchecked") public void channelRead(ChannelHandlerContext ctx, Object msg) { //該channel為客戶端接入時創建的channel final Channel child = (Channel) msg; //添加childHandler child.pipeline().addLast(childHandler); //設置TCP相關屬性:childOptions setChannelOptions(child, childOptions, logger); //設置自定義屬性:childAttrs for (Entry , Object> e: childAttrs) { child.attr((AttributeKey ) e.getKey()).set(e.getValue()); } try { //選擇NioEventLoop并注冊Selector 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); } } //省略了一些代碼。。。。。 }
客戶端channel的pipeline添加childHandler,在服務端EchoServer創建流程中,childHandler的時候,使用了ChannelInitializer的一個自定義實例。并且覆蓋了其initChannel方法,改方法獲取到pipeline并添加具體的Handler。查看ChannelInitializer具體的添加邏輯,handlerAdded方法。其實在initChannel邏輯中,首先是回調到用戶代碼執行initChannel,用戶代碼執行添加Handler的添加操作,之后將ChannelInitializer自己從pipeline中刪除。
public abstract class ChannelInitializerextends ChannelInboundHandlerAdapter { @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { if (ctx.channel().isRegistered()) { // This should always be true with our current DefaultChannelPipeline implementation. // The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering // surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers // will be added in the expected order. //初始化Channel if (initChannel(ctx)) { // We are done with init the Channel, removing the initializer now. removeState(ctx); } } } private boolean initChannel(ChannelHandlerContext ctx) throws Exception { if (initMap.add(ctx)) { // Guard against re-entrance. try { //回調到用戶代碼 initChannel((C) ctx.channel()); } catch (Throwable cause) { // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...). // We do so to prevent multiple calls to initChannel(...). exceptionCaught(ctx, cause); } finally { ChannelPipeline pipeline = ctx.pipeline(); if (pipeline.context(this) != null) { //刪除本身 pipeline.remove(this); } } return true; } return false; } }
2)、設置客戶端TCP相關屬性childOptions和自定義屬性childAttrs
這點在ServerBootstrapAcceptor#init()方法中已經體現
3)、workGroup選擇NioEventLoop并注冊Selector
這要從AbstractBootstrap#initAndRegister()方法開始,然后跟蹤源碼會來到AbstractUnsafe#register()方法
protected abstract class AbstractUnsafe implements Unsafe { //省略了一些代碼。。。。。 @Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } if (isRegistered()) { promise.setFailure(new IllegalStateException("registered to an event loop already")); return; } if (!isCompatible(eventLoop)) { promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; } AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop()) { register0(promise); } else { try { eventLoop.execute(new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { logger.warn( "Force-closing a channel whose registration task was not accepted by an event loop: {}", AbstractChannel.this, t); closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } } } //省略了一些代碼。。。。。 }
最后調用AbstractNioUnsafe#doRegister()方法通過jdk的javaChannel().register完成注冊功能。
protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe { //省略了一些代碼。。。。。 @Override protected void doRegister() throws Exception { boolean selected = false; for (;;) { try { selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this); return; } catch (CancelledKeyException e) { if (!selected) { // Force the Selector to select now as the "canceled" SelectionKey may still be // cached and not removed because no Select.select(..) operation was called yet. eventLoop().selectNow(); selected = true; } else { // We forced a select operation on the selector before but the SelectionKey is still cached // for whatever reason. JDK bug ? throw e; } } } } //省略了一些代碼。。。。。 }4.向Selector注冊讀事件
a)、入口:ServerBootstrap.ServerBootstrapAcceptor#channelRead()#childGroup.register();
public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; child.pipeline().addLast(childHandler); setChannelOptions(child, childOptions, logger); for (Entry, Object> e: childAttrs) { child.attr((AttributeKey ) 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); } }
b)、實際上調用了AbstractChannel.AbstractUnsafe#register0(),觸發了通道激活事件;
//觸發通道激活事件,調用HeadContent的 pipeline.fireChannelActive();
c)、pipeline的頭部開始,即DefaultChannelPipeline.HeadContext#channelActive()從而觸發了readIfIsAutoRead();
@Override public void channelActive(ChannelHandlerContext ctx) { ctx.fireChannelActive(); readIfIsAutoRead(); }
d)、讀事件將從尾部的TailContent#read()被觸發,從而依次執行ctx.read(),從尾部開始,每個outboundHandler的read()事件都被觸發。直到頭部。
@Override public final ChannelPipeline read() { tail.read(); return this; } @Override public ChannelHandlerContext read() { //獲取最近的outboundhandler final AbstractChannelHandlerContext next = findContextOutbound(); EventExecutor executor = next.executor(); //并依次執行其read方法 if (executor.inEventLoop()) { next.invokeRead(); } else { Tasks tasks = next.invokeTasks; if (tasks == null) { next.invokeTasks = tasks = new Tasks(next); } executor.execute(tasks.invokeReadTask); } return this; }
e)、進入頭部HeadContext#read(),并且最終更改了selectionKey,向selector注冊了讀事件
HeadContext#read()
@Override public void read(ChannelHandlerContext ctx) { unsafe.beginRead(); }
AbstractChannel#beginRead()
@Override public final void beginRead() { assertEventLoop(); if (!isActive()) { return; } try { doBeginRead(); } catch (final Exception e) { invokeLater(new Runnable() { @Override public void run() { pipeline.fireExceptionCaught(e); } }); close(voidPromise()); } }
AbstractNioMessageChannel#doBeginRead
@Override protected void doBeginRead() throws Exception { if (inputShutdown) { return; } super.doBeginRead(); }
AbstractNioChannel#doBeginRead()
@Override protected void doBeginRead() throws Exception { // Channel.read() or ChannelHandlerContext.read() was called final SelectionKey selectionKey = this.selectionKey; if (!selectionKey.isValid()) { return; } readPending = true; final int interestOps = selectionKey.interestOps(); if ((interestOps & readInterestOp) == 0) { selectionKey.interestOps(interestOps | readInterestOp); } }
參考文章:
Jorgezhong
Netty如何接入新連接基本流程如上所述,如果有誤,還望各位指正。建議先從前兩篇看起比較好理解點。
【Netty】服務端和客戶端
學習NioEventLoop
如果對 Java、大數據感興趣請長按二維碼關注一波,我會努力帶給你們價值。覺得對你哪怕有一丁點幫助的請幫忙點個贊或者轉發哦。
關注公眾號【愛編碼】,回復2019有相關資料哦。
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/74852.html
摘要:是一個持久化的協議,相對于這種非持久的協議來說。最大的特點就是實現全雙工通信客戶端能夠實時推送消息給服務端,服務端也能夠實時推送消息給客戶端。參考鏈接知乎問題原理原理知乎問題編碼什么用如果文章有錯的地方歡迎指正,大家互相交流。 前言 今天在慕課網上看到了Java的新教程(Netty入門之WebSocket初體驗):https://www.imooc.com/learn/941 WebS...
時間:2018年04月11日星期三 說明:本文部分內容均來自慕課網。@慕課網:https://www.imooc.com 教學源碼:https://github.com/zccodere/s... 學習源碼:https://github.com/zccodere/s... 第一章:課程介紹 1-1 課程介紹 什么是Netty 高性能、事件驅動、異步非阻塞的IO Java開源框架 基于NIO的客戶...
摘要:抽象在中步驟監聽端口對應就是,即事件循環,這里的循環包括兩個部分,一個是新連接的接入,而另一個則是當前存在連接的數據流的讀寫。對的抽象服務端接收數據流的載體都是基于,封裝了許多高可用的,我們可以基于這些與底層數據流做通信。 傳統socket 首先還是先了解下傳統socket下的通信流程 showImg(https://segmentfault.com/img/bVbhjv4?w=842...
摘要:的選擇器允許單個線程監視多個輸入通道。一旦執行的線程已經超過讀取代碼中的某個數據片段,該線程就不會在數據中向后移動通常不會。 1、引言 很多初涉網絡編程的程序員,在研究Java NIO(即異步IO)和經典IO(也就是常說的阻塞式IO)的API時,很快就會發現一個問題:我什么時候應該使用經典IO,什么時候應該使用NIO? 在本文中,將嘗試用簡明扼要的文字,闡明Java NIO和經典IO之...
閱讀 3296·2021-11-24 09:39
閱讀 2805·2021-10-12 10:20
閱讀 1908·2019-08-30 15:53
閱讀 3076·2019-08-30 14:14
閱讀 2600·2019-08-29 15:36
閱讀 1121·2019-08-29 14:11
閱讀 1956·2019-08-26 13:51
閱讀 3408·2019-08-26 13:23