Netty源碼分析——flush流程

前言

承接上篇寫流程随抠，這篇看下flush流程均芽。之前文章中我們已經(jīng)提到過，writeAndFlush操作實(shí)際上是通過pipeline分別進(jìn)行了write和flush操作。具體我們就不看了恕稠，我們直接看下flush。

flush

flush操作同樣是通過pipeline最終傳遞給HeadContext：unsafe.flush();：

123456789101112public final void flush() { //確保不是外部調(diào)用 assertEventLoop(); ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { return; } //添加flush節(jié)點(diǎn) outboundBuffer.addFlush(); //把節(jié)點(diǎn)里的數(shù)據(jù)寫到socket里 flush0();}

最主要的其實(shí)是兩個(gè)步驟樊诺，上文已經(jīng)標(biāo)注了仗考，一個(gè)就是添加flush節(jié)點(diǎn)，一個(gè)就是真正的寫操作词爬。

添加flush節(jié)點(diǎn)

追進(jìn)去看下：

1234567891011121314151617public void addFlush() { Entry entry = unflushedEntry; if (entry != null) { if (flushedEntry == null) { flushedEntry = entry; } do { flushed ++; if (!entry.promise.setUncancellable()) { int pending = entry.cancel(); decrementPendingOutboundBytes(pending, false, true); } entry = entry.next; } while (entry != null); unflushedEntry = null; }}

我們按照上篇文章的狀態(tài)來說秃嗜，當(dāng)前調(diào)用了兩次write的狀態(tài)是這樣的：

調(diào)用完addFlush之后是這樣的：

看到了嗎，實(shí)際上是把flushedEntry和unFlushedEntry交換了一下顿膨。

再假設(shè)一下锅锨，如果我們?cè)谡{(diào)用flush之前調(diào)用了三次write，再調(diào)用flush恋沃，鏈表是這樣的：

添加節(jié)點(diǎn)之后會(huì)繼續(xù)執(zhí)行flush0：

123if (!isFlushPending()) { super.flush0();}

看下這個(gè)isFlushPending：

12SelectionKey selectionKey = selectionKey();return selectionKey.isValid() && (selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0;

這里其實(shí)是在校驗(yàn)必搞，當(dāng)前這個(gè)Channel是否有OP_WRITE，如果當(dāng)前selectionKey是寫事件囊咏，說明有線程執(zhí)行flush過程恕洲，結(jié)合上面的一句：!isFlushPending()，說明如果有線程在進(jìn)行flush過程梅割，就直接返回霜第。

這里其實(shí)我還沒吃透，我的疑問是這里是否有必要進(jìn)行!isFlushPending()判斷炮捧。因?yàn)槲覀冎耙呀?jīng)說過了庶诡，任何flush操作開頭都進(jìn)行了一個(gè)校驗(yàn)：assertEventLoop()，說白了咆课，只有Reactor線程可以調(diào)用flush末誓，那么當(dāng)前線程在執(zhí)行的時(shí)候，怎么可能有別的線程進(jìn)行了flush操作呢书蚪？

這個(gè)問題我會(huì)去debug一下喇澡，然后求證一下作者，有結(jié)果了會(huì)在文章后面加上殊校。

繼續(xù)看晴玖，如果當(dāng)前channel沒有被其他線程操作，這里會(huì)調(diào)用super.flush0为流，回到io.netty.channel.AbstractChannel.AbstractUnsafe#flush0里：

12345678910111213141516171819202122232425262728293031323334353637383940if (inFlush0) { // 防止重復(fù)調(diào)用 return;}// 如果沒有數(shù)據(jù)要flush就返回final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;if (outboundBuffer == null || outboundBuffer.isEmpty()) { return;}inFlush0 = true;// 如果channel失效呕屎，把所有待刷的數(shù)據(jù)設(shè)置為失敗if (!isActive()) { try { if (isOpen()) { outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true); } else { outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } } finally { inFlush0 = false; } return;}try { // 真正的寫操作 doWrite(outboundBuffer);} catch (Throwable t) { if (t instanceof IOException && config().isAutoClose()) { close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } else { try { shutdownOutput(voidPromise(), t); } catch (Throwable t2) { close(voidPromise(), t2, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } }} finally { inFlush0 = false;}

繼續(xù)看doWrite操作，這里會(huì)直接走到io.netty.channel.socket.nio.NioSocketChannel#doWrite里：

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849SocketChannel ch = javaChannel();// 獲取循環(huán)次數(shù)敬察，相當(dāng)于一個(gè)自旋秀睛，保證寫成功int writeSpinCount = config().getWriteSpinCount();do { // 如果buffer里空的，則清理OP_WRITE莲祸，防止Reacotr線程再次處理這個(gè)Channel if (in.isEmpty()) { clearOpWrite(); return; } // 聚合 int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite(); ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite); int nioBufferCnt = in.nioBufferCount(); switch (nioBufferCnt) { case 0: writeSpinCount -= doWrite0(in); break; case 1: { // 如果只有一個(gè)buffer的情況下蹂安，直接把這個(gè)buffer寫進(jìn)去 ByteBuffer buffer = nioBuffers[0]; int attemptedBytes = buffer.remaining(); final int localWrittenBytes = ch.write(buffer); if (localWrittenBytes <= 0) { incompleteWrite(true); return; } adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes, maxBytesPerGatheringWrite); in.removeBytes(localWrittenBytes); --writeSpinCount; break; } default: { // 多個(gè)buffer的情況下椭迎，寫nioBufferCnt個(gè)buffer進(jìn)去 long attemptedBytes = in.nioBufferSize(); final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt); if (localWrittenBytes <= 0) { incompleteWrite(true); return; } adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes, maxBytesPerGatheringWrite); in.removeBytes(localWrittenBytes); --writeSpinCount; break; } }} while (writeSpinCount > 0);// 是否寫完成incompleteWrite(writeSpinCount < 0);

上面的過程中，我只是粗略的寫了一下過程田盈，其實(shí)里面的細(xì)節(jié)非常多畜号，我們一點(diǎn)一點(diǎn)來看。

先看著幾句：

123int maxBytesPerGatheringWrite = ((NioSocketChannelConfig) config).getMaxBytesPerGatheringWrite();ByteBuffer[] nioBuffers = in.nioBuffers(1024, maxBytesPerGatheringWrite);int nioBufferCnt = in.nioBufferCount();

首先獲取聚合寫的最大字節(jié)數(shù)允瞧，聚合寫简软，在原生NIO的概念中，是把幾個(gè)Buffer的數(shù)據(jù)寫到一個(gè)Channel里瓷式，就是說一次最多寫多少字節(jié)數(shù)據(jù)替饿。然后進(jìn)入到nioBuffers方法，這個(gè)方法做什么注釋上有說：如果緩沖區(qū)里全都是ByteBuf贸典，則返回直接NIO緩沖區(qū)的Buffer數(shù)組（其實(shí)就是把ByteBuf里的數(shù)據(jù)寫到原生Buffer里）视卢，nioBufferCount和nioBufferSize分別代表返回?cái)?shù)組中原生NIO Buffer的數(shù)量和NIO緩沖區(qū)的可讀字節(jié)總數(shù)±韧眨看代碼据过，有點(diǎn)長拆開看：

123456789long nioBufferSize = 0;int nioBufferCount = 0;final InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();ByteBuffer[] nioBuffers = NIO_BUFFERS.get(threadLocalMap);Entry entry = flushedEntry;while (isFlushedEntry(entry) && entry.msg instanceof ByteBuf) { //...}return nioBuffers;

先是從當(dāng)前線程里獲取ByteBuffer，這里可以看出妒挎，其實(shí)ByteBuffer是被緩存了的（如果沒有創(chuàng)建一個(gè)長度為1024的ByteBuffer數(shù)組）绳锅，不需要每次創(chuàng)建。

然后是循環(huán)所有的flushedEntry酝掩，這里鳞芙，我們回顧一下上面addFlush之后的圖，其實(shí)循環(huán)中會(huì)不停地把flushedEntry前移期虾，直到flushedEntry和tailEntry中的節(jié)點(diǎn)全部都被處理原朝。isFlushEntry的代碼：e != null && e != unflushedEntry;，其實(shí)就是镶苞，不是unflushedEntry的都是flushedEntry喳坠。

這里我們可以看到，另外一個(gè)循環(huán)的條件就是entry.msg instanceof ByteBuf茂蚓，說明這個(gè)方法只處理ByteBuf壕鹉。

繼續(xù)看循環(huán)里：

1234567891011121314151617181920212223ByteBuf buf = (ByteBuf) entry.msg;final int readerIndex = buf.readerIndex();final int readableBytes = buf.writerIndex() - readerIndex;// 如果有數(shù)據(jù)if (readableBytes > 0) { if (maxBytes - readableBytes < nioBufferSize && nioBufferCount != 0) { break; } // NIO Buffer可讀字節(jié)數(shù)+ByteBuf的可讀字節(jié)數(shù) nioBufferSize += readableBytes; int count = entry.count; if (count == -1) { // 初始化entry的count entry.count = count = buf.nioBufferCount(); } int neededSpace = min(maxCount, nioBufferCount + count); if (neededSpace > nioBuffers.length) { //如果實(shí)際需要的空間，比之前得到的ByteBuffer數(shù)組數(shù)量大聋涨，就擴(kuò)容晾浴，然后緩存起來 nioBuffers = expandNioBufferArray(nioBuffers, neededSpace, nioBufferCount); NIO_BUFFERS.set(threadLocalMap, nioBuffers); }}

這里說下這個(gè)流程，聲明一下牍白，ByteBuffer指的是原生Buffer脊凰，ByteBuf是Netty自己封裝的Buffer。

先拿出來一個(gè)Entry的ByteBuf淹朋，看看可讀的字節(jié)多少笙各。然后初始化entry的count，用的是組成這個(gè)ByteBuf的ByteBuffer（原生）的數(shù)量（這里這個(gè)nioBufferCount大部分情況下返回1础芍，及一個(gè)ByteBuf對(duì)應(yīng)一個(gè)ByteBuffer）杈抢。

然后看看需要的空間是多少，需要的空間默認(rèn)情況下是1024和已有需要的ByteBuffer數(shù)量+count（及ByteBuf.nioBufferCount）二者之間的最小值仑性，及惶楼，最多搞1024個(gè)ByteBuffer。

然后對(duì)比需要空間和提供空間诊杆，及對(duì)比之前分配的ByteBuffer[]的length和需要的ByteBuffer的數(shù)量歼捐，如果需要的空間大，就擴(kuò)容（這里可以類比一下ArrayList的擴(kuò)容）晨汹。也就是豹储，不管怎么樣，都會(huì)分配足夠的ByteBuffer使用淘这。

可能這么看起來有點(diǎn)繞剥扣，我舉個(gè)例子：

假設(shè)我們有八個(gè)節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)的ByteBuf在Flush的時(shí)候铝穷，數(shù)據(jù)都會(huì)寫入到1個(gè)ByteBuffer里钠怯，然后我們開始循環(huán)這個(gè)八個(gè)節(jié)點(diǎn)，循環(huán)之前我記錄一下一共需要多少個(gè)ByteBuffer數(shù)組（比如叫count曙聂，循環(huán)前就是0）晦炊，然后我們有一個(gè)分配給我們的ByteBuffer數(shù)組（比如叫fenpei）。

代碼應(yīng)該是這樣的：

123456789101112// 循環(huán)八個(gè)節(jié)點(diǎn)int count = 0;for (Entry e : entries) { ByteBuf b = e.msg; // 這里大部分ByteBuf會(huì)返回1 int c = b.nioBufferCount(); count += c; if (count > fenpei.length) { //擴(kuò)容 expandNioBufferArray(fenpei) }}

這樣就比較直接了宁脊，再看不懂的断国。。朦佩。emmm并思，哈哈哈哈哈~

繼續(xù)看：

12345678910111213141516171819202122232425if (count == 1) { // 1個(gè)ByteBuff對(duì)應(yīng)1個(gè)ByteBuffer ByteBuffer nioBuf = entry.buf; if (nioBuf == null) { // 初始化ByteBuffer， entry.buf = nioBuf = buf.internalNioBuffer(readerIndex, readableBytes); } // 放到數(shù)組里 nioBuffers[nioBufferCount++] = nioBuf;} else { // 一個(gè)ByteBuf對(duì)應(yīng)多個(gè)ByteBuffer语稠，初始化多個(gè)ByteBuffer宋彼，循環(huán)放到數(shù)組里 ByteBuffer[] nioBufs = entry.bufs; if (nioBufs == null) { entry.bufs = nioBufs = buf.nioBuffers(); } for (int i = 0; i < nioBufs.length && nioBufferCount < maxCount; ++i) { ByteBuffer nioBuf = nioBufs[i]; if (nioBuf == null) { break; } else if (!nioBuf.hasRemaining()) { continue; } nioBuffers[nioBufferCount++] = nioBuf; }}

這個(gè)方法我們總結(jié)一下，就是分配數(shù)組仙畦，這個(gè)數(shù)組一開始會(huì)初始化一個(gè)長度為1024的ByteBuffer數(shù)組输涕，如果不夠用就擴(kuò)容，里面的ByteBuffer能容納的數(shù)據(jù)慨畸，對(duì)應(yīng)每個(gè)節(jié)點(diǎn)ByteBuf里有的數(shù)據(jù)莱坎。這個(gè)地方其實(shí)并沒有看到ByteBuf向?qū)?yīng)的ByteBuffer里寫數(shù)據(jù)的地方，關(guān)于這個(gè)問題寸士，大家可以跟一下buf.internalNioBuffer(readerIndex, readableBytes)這里檐什，這里是會(huì)把數(shù)據(jù)搞到ByteBuffer里的碴卧。

至此分配就結(jié)束了，然后繼續(xù)往下看doWrite乃正，接下來進(jìn)入了一個(gè)switch住册，條件就是有多少個(gè)原生ByteBuffer要寫，我們看看default的情況：

12345678910111213141516// 整個(gè)Entry鏈的可讀數(shù)據(jù)long attemptedBytes = in.nioBufferSize();// 向管道中寫數(shù)據(jù)final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);// 如果寫失敗了瓮具，這個(gè)要細(xì)說if (localWrittenBytes <= 0) { incompleteWrite(true); return;}// 調(diào)整最大聚合寫的字節(jié)數(shù)adjustMaxBytesPerGatheringWrite((int) attemptedBytes, (int) localWrittenBytes, maxBytesPerGatheringWrite);// remove節(jié)點(diǎn) in.removeBytes(localWrittenBytes);// 自旋次數(shù)-1--writeSpinCount;

向channel中寫入數(shù)據(jù)荧飞，如果localWrittenBytes < 0，這里是說明這個(gè)Channel不可用名党，其實(shí)就是寫失敗了（或者說沒有全部寫到Channel里叹阔？這個(gè)地方存疑，我沒有驗(yàn)證過）传睹。

這里寫失敗了怎么辦耳幢，我們看下incompleteWrite，注意入?yún)⑹莟rue蒋歌，下面會(huì)說到：

1234567891011121314151617// 這里這個(gè)setOpWrite就是入?yún)rueif (setOpWrite) { setOpWrite();} else { clearOpWrite(); eventLoop().execute(flushTask);}setOpWrite方法：final SelectionKey key = selectionKey();if (!key.isValid()) { return;}final int interestOps = key.interestOps();if ((interestOps & SelectionKey.OP_WRITE) == 0) { key.interestOps(interestOps | SelectionKey.OP_WRITE);}

如果失敗了帅掘，就去setOpWrite，其實(shí)就是給Channel注冊(cè)了一個(gè)OP_WRITE堂油，然后就return了修档。

這里為什么要打上OP_WRITE呢，打上有什么用呢府框？還記得上文中我提出的問題么吱窝，關(guān)于為什么flush0之前要進(jìn)行判斷isFlushPending，這里讓我細(xì)細(xì)說來（為什么判斷isFlushPending的疑惑已經(jīng)解開了迫靖，這篇文章拖了幾天院峡，這幾天中來來回回的看源碼、請(qǐng)教大神系宜、debug照激。。終于有所突破）盹牧。

先放下isFlushPending俩垃，繼續(xù)說這個(gè)OP_WRITE。這里還記得什么時(shí)候打上OP_WRITE么汰寓，是Channel寫失敗的時(shí)候口柳！我們先不管這個(gè)OP_WRITE的具體含義，就認(rèn)為是一個(gè)標(biāo)記有滑，標(biāo)記這個(gè)管道不可用跃闹，這時(shí)候，請(qǐng)問：管道不可用的情況下，如果我還想進(jìn)行Flush操作望艺，即向管道中寫數(shù)據(jù)苛秕，這時(shí)候能成功么？答案是不行找默！怎么優(yōu)化想帅？太簡(jiǎn)單了，提前返回就可以了啡莉，每次Flush的時(shí)候先看看管道是否可用！

到這旨剥，isFlushPending的作用就體現(xiàn)出來了咧欣！OP_WRITE我們就把它當(dāng)成一個(gè)普通的標(biāo)記，如果Channel上有這個(gè)標(biāo)記轨帜，就表示不可寫魄咕。

那么為什么用OP_WRITE標(biāo)識(shí)不可寫呢，命名OP_WRITE的含義就是可寫鞍龈浮哮兰！這里就要說回到Reacotr要做的三件事中的處理select到的事件了，看一下processSelectedKey：

1234if ((readyOps & SelectionKey.OP_WRITE) != 0) { // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write ch.unsafe().forceFlush();}

其中有這么一句苟弛，這里的注釋我也抄過來了喝滞，就是說，如果輪訓(xùn)到寫事件膏秫，就去執(zhí)行forceFlush右遭，然后清理掉OP_WRITE。

回想我們寫失敗的地方缤削。如果寫失敗窘哈，就給管道注冊(cè)O(shè)P_WRITE，然后Reacotr線程會(huì)不斷地去select亭敢，一旦Channel可用滚婉，那么這個(gè)Channel由于之前注冊(cè)了OP_WRITE，就會(huì)被Reactor線程select出來帅刀，然后進(jìn)行forceFlush让腹，這個(gè)forceFlush其實(shí)就是調(diào)用了flush0，重新走了一遍flush操作劝篷，注意兩個(gè)地方：

forceFlush不會(huì)添加flush節(jié)點(diǎn)（不會(huì)調(diào)用addFlush）

forceFlush不會(huì)進(jìn)行isFlushPending校驗(yàn)

為什么不進(jìn)行isFlushPending：我們說過了哨鸭，OP_WRITE的意思是管道不可用，那么被select出來的管道一定是可用的娇妓，直接進(jìn)行寫操作像鸡。

這樣我們就理順了寫失敗的流程：

如果寫失敗，給管道注冊(cè)一個(gè)OP_WRITE

其他的flush操作都會(huì)直接返回（被isFlushPending）攔截

管道可用后，被Reacotr線程select到只估，進(jìn)行forceFlush操作

收一下志群，看回到doWrite方法里，如果寫成功蛔钙，進(jìn)行in.removeBytes(localWrittenBytes);操作锌云，remove掉這個(gè)節(jié)點(diǎn)。

注意吁脱，正常情況下桑涎，結(jié)束doWrite操作是在：

1234if (in.isEmpty()) { clearOpWrite(); return;}

這里返回的。如果樂觀鎖的默認(rèn)16次都循環(huán)完兼贡，操作還沒結(jié)束攻冷，又會(huì)進(jìn)行incompleteWrite(writeSpinCount < 0)操作，如果執(zhí)行了16次循環(huán)以后遍希，ChannelOutboundBuffer中還有Entry等曼，writeSpinCount < 0成立，設(shè)置一個(gè)OP_WRITE凿蒜，然后等著被Reacotr線程select禁谦。如果大于0，這種情況比較特殊废封，寫入的ByteBuf或者FileRegion只有一個(gè)州泊，但是這個(gè)ByteBuf是不可讀的，或者region.transferred() >= region.count()漂洋。這時(shí)候會(huì)走到incompleteWrite(false)里拥诡，這里執(zhí)行clearOpWrite();和eventLoop().execute(flushTask);，清理掉OP_WRITE讓通道繼續(xù)可寫氮发，然后再次扔了一個(gè)flushTask到NioEventLoop里渴肉，這里其是讓出資源，讓Reacotr可以處理其他的task爽冕。至此仇祭，整個(gè)寫流程就結(jié)束了，寫的比較細(xì)颈畸，大家多多琢磨乌奇，多多思考，才會(huì)有更多收獲眯娱！

如果想學(xué)習(xí)Java工程化礁苗、高性能及分布式、深入淺出徙缴。微服務(wù)试伙、Spring，MyBatis，Netty源碼分析的朋友可以加我的Java進(jìn)階群：617434785疏叨，群里有阿里大牛直播講解技術(shù)潘靖，以及Java大型互聯(lián)網(wǎng)技術(shù)的視頻免費(fèi)分享給大家。