作者 | nxlhero 來(lái)源 | https://blog.51cto.com/nxlhero/2515849

一、背景

生產(chǎn)擁堵回顧

近期在一次生產(chǎn)發(fā)布過(guò)程中，因?yàn)橥话l(fā)的流量，出現(xiàn)了擁堵。系統(tǒng)的部署圖如下，客戶端通過(guò)Http協(xié)議訪問(wèn)到Dubbo的消費(fèi)者，消費(fèi)者通過(guò)Dubbo協(xié)議訪問(wèn)服務(wù)提供者。這是單個(gè)機(jī)房，8個(gè)消費(fèi)者3個(gè)提供者，共兩個(gè)機(jī)房對(duì)外服務(wù)。

在發(fā)布的過(guò)程中，摘掉一個(gè)機(jī)房，讓另一個(gè)機(jī)房對(duì)外服務(wù)，然后摘掉的機(jī)房發(fā)布新版本，然后再互換，最終兩個(gè)機(jī)房都以新版本對(duì)外服務(wù)。問(wèn)題就出現(xiàn)單機(jī)房對(duì)外服務(wù)的時(shí)候，這時(shí)候單機(jī)房還是老版本應(yīng)用。以前不知道晚上會(huì)有一個(gè)高峰，結(jié)果當(dāng)晚的高峰和早上的高峰差不多了，單機(jī)房扛不住這么大的流量，出現(xiàn)了擁堵。這些流量的特點(diǎn)是并發(fā)比較高，個(gè)別交易返回報(bào)文較大，因?yàn)槭且粋€(gè)產(chǎn)品列表頁(yè)，點(diǎn)擊后會(huì)發(fā)送多個(gè)交易到后臺(tái)。

在問(wèn)題發(fā)生時(shí)，因?yàn)椴磺宄顟B(tài)，先切到另外一個(gè)機(jī)房，結(jié)果也擁堵了，最后整體回退，折騰了一段時(shí)間沒有問(wèn)題了。當(dāng)時(shí)有一些現(xiàn)象：

(1)提供者的CPU內(nèi)存等都不高，第一個(gè)機(jī)房的最高CPU 66%(8核虛擬機(jī))，第二個(gè)機(jī)房的最高CPU 40%(16核虛擬機(jī))。消費(fèi)者的最高CPU只有30%多(兩個(gè)消費(fèi)者結(jié)點(diǎn)位于同一臺(tái)虛擬機(jī)上)

(2)在擁堵的時(shí)候，服務(wù)提供者的Dubbo業(yè)務(wù)線程池(下面會(huì)詳細(xì)介紹這個(gè)線程池)并沒滿，最多到了300，最大值是500。但是把這個(gè)機(jī)房摘下后，也就是沒有外部的流量了，線程池反而滿了，而且好幾分鐘才把堆積的請(qǐng)求處理完。

(3)通過(guò)監(jiān)控工具統(tǒng)計(jì)的每秒進(jìn)入Dubbo業(yè)務(wù)線程池的請(qǐng)求數(shù)，在擁堵時(shí)，時(shí)而是0，時(shí)而特別大，在日間正常的時(shí)候，這個(gè)值不存在為0的時(shí)候。

事故原因猜測(cè)

當(dāng)時(shí)其他指標(biāo)沒有檢測(cè)到異常，也沒有打Dump，我們通過(guò)分析這些現(xiàn)象以及我們的Dubbo配置，猜測(cè)是在網(wǎng)絡(luò)上發(fā)生了擁堵，而影響擁堵的關(guān)鍵參數(shù)就是Dubbo協(xié)議的連接數(shù)，我們默認(rèn)使用了單個(gè)連接，但是消費(fèi)者數(shù)量較少，沒能充分把網(wǎng)絡(luò)資源利用起來(lái)。

默認(rèn)的情況下，每個(gè)Dubbo消費(fèi)者與Dubbo提供者建立一個(gè)長(zhǎng)連接，Dubbo官方對(duì)此的建議是：

Dubbo 缺省協(xié)議采用單一長(zhǎng)連接和 NIO 異步通訊，適合于小數(shù)據(jù)量大并發(fā)的服務(wù)調(diào)用，以及服務(wù)消費(fèi)者機(jī)器數(shù)遠(yuǎn)大于服務(wù)提供者機(jī)器數(shù)的情況。

反之，Dubbo 缺省協(xié)議不適合傳送大數(shù)據(jù)量的服務(wù)，比如傳文件，傳視頻等，除非請(qǐng)求量很低。

(http://dubbo.apache.org/zh-cn/docs/user/references/protocol/dubbo.html)

以下也是Dubbo官方提供的一些常見問(wèn)題回答：

為什么要消費(fèi)者比提供者個(gè)數(shù)多?

因 dubbo 協(xié)議采用單一長(zhǎng)連接，假設(shè)網(wǎng)絡(luò)為千兆網(wǎng)卡，根據(jù)測(cè)試經(jīng)驗(yàn)數(shù)據(jù)每條連接最多只能壓滿 7MByte(不同的環(huán)境可能不一樣，供參考)，理論上 1 個(gè)服務(wù)提供者需要 20 個(gè)服務(wù)消費(fèi)者才能壓滿網(wǎng)卡。

為什么不能傳大包?

因 dubbo 協(xié)議采用單一長(zhǎng)連接，如果每次請(qǐng)求的數(shù)據(jù)包大小為 500KByte，假設(shè)網(wǎng)絡(luò)為千兆網(wǎng)卡，每條連接最大 7MByte(不同的環(huán)境可能不一樣，供參考)，單個(gè)服務(wù)提供者的 TPS(每秒處理事務(wù)數(shù))最大為：128MByte / 500KByte = 262。單個(gè)消費(fèi)者調(diào)用單個(gè)服務(wù)提供者的 TPS(每秒處理事務(wù)數(shù))最大為：7MByte / 500KByte = 14。如果能接受，可以考慮使用，否則網(wǎng)絡(luò)將成為瓶頸。

為什么采用異步單一長(zhǎng)連接?

因?yàn)榉?wù)的現(xiàn)狀大都是服務(wù)提供者少，通常只有幾臺(tái)機(jī)器，而服務(wù)的消費(fèi)者多，可能整個(gè)網(wǎng)站都在訪問(wèn)該服務(wù)，比如 Morgan 的提供者只有 6 臺(tái)提供者，卻有上百臺(tái)消費(fèi)者，每天有 1.5 億次調(diào)用，如果采用常規(guī)的 hessian 服務(wù)，服務(wù)提供者很容易就被壓跨，通過(guò)單一連接，保證單一消費(fèi)者不會(huì)壓死提供者，長(zhǎng)連接，減少連接握手驗(yàn)證等，并使用異步 IO，復(fù)用線程池，防止 C10K 問(wèn)題。

因?yàn)槲覀兊南M(fèi)者數(shù)量和提供者數(shù)量都不多，所以很可能是連接數(shù)不夠，導(dǎo)致網(wǎng)絡(luò)傳輸出現(xiàn)了瓶頸。以下我們通過(guò)詳細(xì)分析Dubbo協(xié)議和一些實(shí)驗(yàn)來(lái)驗(yàn)證我們的猜測(cè)。

二、Dubbo通信流程詳解

我們用的Dubbo版本比較老，是2.5.x的，它使用的netty版本是3.2.5，最新版的Dubbo在線程模型上有一些修改，我們以下的分析是以2.5.10為例。

以圖和部分代碼說(shuō)明Dubbo協(xié)議的調(diào)用過(guò)程，代碼只寫了一些關(guān)鍵部分，使用的是netty3，dubbo線程池?zé)o隊(duì)列，同步調(diào)用，以下代碼包含了Dubbo和Netty的代碼。

整個(gè)Dubbo一次調(diào)用過(guò)程如下：

1.請(qǐng)求入隊(duì)

我們通過(guò)Dubbo調(diào)用一個(gè)rpc服務(wù)，調(diào)用線程其實(shí)是把這個(gè)請(qǐng)求封裝后放入了一個(gè)隊(duì)列里。這個(gè)隊(duì)列是netty的一個(gè)隊(duì)列，這個(gè)隊(duì)列的定義如下，是一個(gè)Linked隊(duì)列，不限長(zhǎng)度。

class?NioWorker?implements?Runnable?{
????...
????private?final?Queue?writeTaskQueue?=?new?LinkedTransferQueue();
????...
}

主線程經(jīng)過(guò)一系列調(diào)用，最終通過(guò)NioClientSocketPipelineSink類里的方法把請(qǐng)求放入這個(gè)隊(duì)列，放入隊(duì)列的請(qǐng)求，包含了一個(gè)請(qǐng)求ID，這個(gè)ID很重要。

2.調(diào)用線程等待

入隊(duì)后，netty會(huì)返回給調(diào)用線程一個(gè)Future，然后調(diào)用線程等待在Future上。這個(gè)Future是Dubbo定義的，名字叫DefaultFuture，主調(diào)用線程調(diào)用DefaultFuture.get(timeout)，等待通知，所以我們看與Dubbo相關(guān)的ThreadDump，經(jīng)常會(huì)看到線程停在這，這就是在等后臺(tái)返回。

public?class?DubboInvoker<T>?extends?AbstractInvoker<T>?{
????...
???@Override
????protected?Result?doInvoke(final?Invocation?invocation)?throws?Throwable?{
?????????...???
?????????return?(Result)?currentClient.request(inv,?timeout).get();?//currentClient.request(inv,?timeout)返回了一個(gè)DefaultFuture
????}
????...
}

我們可以看一下這個(gè)DefaultFuture的實(shí)現(xiàn)，

public?class?DefaultFuture?implements?ResponseFuture?{

????private?static?final?Map?CHANNELS?=?new?ConcurrentHashMap();
????private?static?final?Map?FUTURES?=?new?ConcurrentHashMap();

????//?invoke?id.
????private?final?long?id;??????//Dubbo請(qǐng)求的id，每個(gè)消費(fèi)者都是一個(gè)從0開始的long類型
????private?final?Channel?channel;
????private?final?Request?request;
????private?final?int?timeout;
????private?final?Lock?lock?=?new?ReentrantLock();
????private?final?Condition?done?=?lock.newCondition();
????private?final?long?start?=?System.currentTimeMillis();
????private?volatile?long?sent;
????private?volatile?Response?response;
????private?volatile?ResponseCallback?callback;
????public?DefaultFuture(Channel?channel,?Request?request,?int?timeout)?{
????????this.channel?=?channel;
????????this.request?=?request;
????????this.id?=?request.getId();
????????this.timeout?=?timeout?>?0???timeout?:?channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY,?Constants.DEFAULT_TIMEOUT);
????????//?put?into?waiting?map.
????????FUTURES.put(id,?this);????//等待時(shí)以id為key把Future放入全局的Future?Map中，這樣回復(fù)數(shù)據(jù)回來(lái)了可以根據(jù)id找到對(duì)應(yīng)的Future通知主線程
????????CHANNELS.put(id,?channel);
????}

3.IO線程讀取隊(duì)列里的數(shù)據(jù)

這個(gè)工作是由netty的IO線程池完成的，也就是NioWorker，對(duì)應(yīng)的類叫NioWorker。它會(huì)死循環(huán)的執(zhí)行select，在select中，會(huì)一次性把隊(duì)列中的寫請(qǐng)求處理完，select的邏輯如下：

public?void?run()?{
????????for?(;;)?{
???????????....
????????????????SelectorUtil.select(selector);

????????????????proce***egisterTaskQueue();?
????????????????processWriteTaskQueue();?//先處理隊(duì)列里的寫請(qǐng)求
????????????????processSelectedKeys(selector.selectedKeys());?//再處理select事件,讀寫都可能有
???????????....
????????}
????}
????private?void?processWriteTaskQueue()?throws?IOException?{
????????for?(;;)?{
????????????final?Runnable?task?=?writeTaskQueue.poll();//這個(gè)隊(duì)列就是調(diào)用線程把請(qǐng)求放進(jìn)去的隊(duì)列
????????????if?(task?==?null)?{
????????????????break;
????????????}
????????????task.run();?//寫數(shù)據(jù)
????????????cleanUpCancelledKeys();
????????}
????}

4.IO線程把數(shù)據(jù)寫到Socket緩沖區(qū)

這一步很重要，跟我們遇到的性能問(wèn)題相關(guān)，還是NioWorker，也就是上一步的task.run()，它的實(shí)現(xiàn)如下：

void?writeFromTaskLoop(final?NioSocketChannel?ch)?{
????????if?(!ch.writeSuspended)?{?//這個(gè)地方很重要，如果writeSuspended了，那么就直接跳過(guò)這次寫
????????????write0(ch);
????????}
????}
????private?void?write0(NioSocketChannel?channel)?{
????????......
????????final?int?writeSpinCount?=?channel.getConfig().getWriteSpinCount();?//netty可配置的一個(gè)參數(shù)，默認(rèn)是16
????????synchronized?(channel.writeLock)?{
????????????channel.inWriteNowLoop?=?true;
????????????for?(;;)?{

????????????????????for?(int?i?=?writeSpinCount;?i?>?0;?i?--)?{?//每次最多嘗試16次
????????????????????????localWrittenBytes?=?buf.transferTo(ch);
????????????????????????if?(localWrittenBytes?!=?0)?{
????????????????????????????writtenBytes?+=?localWrittenBytes;
????????????????????????????break;
????????????????????????}
????????????????????????if?(buf.finished())?{
????????????????????????????break;
????????????????????????}
????????????????????}

????????????????????if?(buf.finished())?{
????????????????????????//?Successful?write?-?proceed?to?the?next?message.
????????????????????????buf.release();
????????????????????????channel.currentWriteEvent?=?null;
????????????????????????channel.currentWriteBuffer?=?null;
????????????????????????evt?=?null;
????????????????????????buf?=?null;
????????????????????????future.setSuccess();
????????????????????}?else?{
????????????????????????//?Not?written?fully?-?perhaps?the?kernel?buffer?is?full.?
????????????????????????//重點(diǎn)在這，如果寫16次還沒寫完，可能是內(nèi)核緩沖區(qū)滿了，writeSuspended被設(shè)置為true
????????????????????????addOpWrite?=?true;
????????????????????????channel.writeSuspended?=?true;
????????????????????????......
????????????????????}
????????????......
????????????if?(open)?{
????????????????if?(addOpWrite)?{
????????????????????setOpWrite(channel);
????????????????}?else?if?(removeOpWrite)?{
????????????????????clearOpWrite(channel);
????????????????}
????????????}
????????????......
????????}

????????fireWriteComplete(channel,?writtenBytes);
????}

正常情況下，隊(duì)列中的寫請(qǐng)求要通過(guò)processWriteTaskQueue處理掉，但是這些寫請(qǐng)求也同時(shí)注冊(cè)到了selector上，如果processWriteTaskQueue寫成功，就會(huì)刪掉selector上的寫請(qǐng)求。如果Socket的寫緩沖區(qū)滿了，對(duì)于NIO，會(huì)立刻返回，對(duì)于BIO，會(huì)一直等待。Netty使用的是NIO，它嘗試16次后，還是不能寫成功，它就把writeSuspended設(shè)置為true，這樣接下來(lái)的所有寫請(qǐng)求都會(huì)被跳過(guò)。那什么時(shí)候會(huì)再寫呢？這時(shí)候就得靠selector了，它如果發(fā)現(xiàn)socket可寫，就把這些數(shù)據(jù)寫進(jìn)去。

下面是processSelectedKeys里寫的過(guò)程，因?yàn)樗前l(fā)現(xiàn)socket可寫才會(huì)寫，所以直接把writeSuspended設(shè)為false。

void?writeFromSelectorLoop(final?SelectionKey?k)?{
????????NioSocketChannel?ch?=?(NioSocketChannel)?k.attachment();
????????ch.writeSuspended?=?false;
????????write0(ch);
????}

5.數(shù)據(jù)從消費(fèi)者的socket發(fā)送緩沖區(qū)傳輸?shù)教峁┱叩慕邮站彌_區(qū)

這個(gè)是操作系統(tǒng)和網(wǎng)卡實(shí)現(xiàn)的，應(yīng)用層的write寫成功了，并不代表對(duì)面能收到，當(dāng)然tcp會(huì)通過(guò)重傳能機(jī)制盡量保證對(duì)端收到。

6.服務(wù)端IO線程從緩沖區(qū)讀取請(qǐng)求數(shù)據(jù)

這個(gè)是服務(wù)端的NIO線程實(shí)現(xiàn)的，在processSelectedKeys中。

public?void?run()?{
???????for?(;;)?{
??????????....
???????????????SelectorUtil.select(selector);

???????????????proce***egisterTaskQueue();?
???????????????processWriteTaskQueue();?
???????????????processSelectedKeys(selector.selectedKeys());?//再處理select事件,讀寫都可能有
??????????....
???????}
???}
???private?void?processSelectedKeys(Set?selectedKeys)?throws?IOException?{
???????for?(Iterator?i?=?selectedKeys.iterator();?i.hasNext();)?{
???????????SelectionKey?k?=?i.next();
???????????i.remove();
???????????try?{
???????????????int?readyOps?=?k.readyOps();
???????????????if?((readyOps?&?SelectionKey.OP_READ)?!=?0?||?readyOps?==?0)?{
???????????????????if?(!read(k))?{
???????????????????????//?Connection?already?closed?-?no?need?to?handle?write.
???????????????????????continue;
???????????????????}
???????????????}
???????????????if?((readyOps?&?SelectionKey.OP_WRITE)?!=?0)?{
???????????????????writeFromSelectorLoop(k);
???????????????}
???????????}?catch?(CancelledKeyException?e)?{
???????????????close(k);
???????????}

???????????if?(cleanUpCancelledKeys())?{
???????????????break;?//?break?the?loop?to?avoid?ConcurrentModificationException
???????????}
???????}
???}
???private?boolean?read(SelectionKey?k)?{
??????......

???????????//?Fire?the?event.
???????????fireMessageReceived(channel,?buffer);??//讀取完后，最終會(huì)調(diào)用這個(gè)函數(shù)，發(fā)送一個(gè)收到信息的事件
??????......

???}

7.IO線程把請(qǐng)求交給Dubbo線程池

按配置不同，走的Handler不同，配置dispatch為all，走的handler如下。下面IO線程直接交給一個(gè)ExecutorService來(lái)處理這個(gè)請(qǐng)求，出現(xiàn)了熟悉的報(bào)錯(cuò)“Threadpool is exhausted"，業(yè)務(wù)線程池滿時(shí)，如果沒有隊(duì)列，就會(huì)報(bào)這個(gè)錯(cuò)。

public?class?AllChannelHandler?extends?WrappedChannelHandler?{
????......
????public?void?received(Channel?channel,?Object?message)?throws?RemotingException?{
????????ExecutorService?cexecutor?=?getExecutorService();
????????try?{
????????????cexecutor.execute(new?ChannelEventRunnable(channel,?handler,?ChannelState.RECEIVED,?message));
????????}?catch?(Throwable?t)?{
????????????//TODO?A?temporary?solution?to?the?problem?that?the?exception?information?can?not?be?sent?to?the?opposite?end?after?the?thread?pool?is?full.?Need?a?refactoring
????????????//fix?The?thread?pool?is?full,?refuses?to?call,?does?not?return,?and?causes?the?consumer?to?wait?for?time?out
????????????if(message?instanceof?Request?&&?t?instanceof?RejectedExecutionException){
????????????????Request?request?=?(Request)message;
????????????????if(request.isTwoWay()){
????????????????????String?msg?=?"Server?side("?+?url.getIp()?+?","?+?url.getPort()?+?")?threadpool?is?exhausted?,detail?msg:"?+?t.getMessage();
????????????????????Response?response?=?new?Response(request.getId(),?request.getVersion());
????????????????????response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
????????????????????response.setErrorMessage(msg);
????????????????????channel.send(response);
????????????????????return;
????????????????}
????????????}
????????????throw?new?ExecutionException(message,?channel,?getClass()?+?"?error?when?process?received?event?.",?t);
????????}
????}
????......
}

8.服務(wù)端Dubbo線程池處理完請(qǐng)求后，把返回報(bào)文放入隊(duì)列

線程池會(huì)調(diào)起下面的函數(shù)

public?class?HeaderExchangeHandler?implements?ChannelHandlerDelegate?{
????......
????Response?handleRequest(ExchangeChannel?channel,?Request?req)?throws?RemotingException?{
????????Response?res?=?new?Response(req.getId(),?req.getVersion());
????????......
????????//?find?handler?by?message?class.
????????Object?msg?=?req.getData();
????????try?{
????????????//?handle?data.
????????????Object?result?=?handler.reply(channel,?msg);???//真正的業(yè)務(wù)邏輯類
????????????res.setStatus(Response.OK);
????????????res.setResult(result);
????????}?catch?(Throwable?e)?{
????????????res.setStatus(Response.SERVICE_ERROR);
????????????res.setErrorMessage(StringUtils.toString(e));
????????}
????????return?res;
????}

????public?void?received(Channel?channel,?Object?message)?throws?RemotingException?{
???????......

????????????if?(message?instanceof?Request)?{
????????????????//?handle?request.
????????????????Request?request?=?(Request)?message;

????????????????????if?(request.isTwoWay())?{
????????????????????????Response?response?=?handleRequest(exchangeChannel,?request);?//處理業(yè)務(wù)邏輯，得到一個(gè)Response
????????????????????????channel.send(response);??//回寫response
????????????????????}?
????????????}
???????......

?}

channel.send(response)最終調(diào)用了NioServerSocketPipelineSink里的方法把返回報(bào)文放入隊(duì)列。

9.服務(wù)端IO線程從隊(duì)列中取出數(shù)據(jù)

與流程3一樣

10.服務(wù)端IO線程把回復(fù)數(shù)據(jù)寫入Socket發(fā)送緩沖區(qū)

IO線程寫數(shù)據(jù)的時(shí)候，寫入到TCP緩沖區(qū)就算成功了。但是如果緩沖區(qū)滿了，會(huì)寫不進(jìn)去。對(duì)于阻塞和非阻塞IO，返回結(jié)果不一樣，阻塞IO會(huì)一直等，而非阻塞IO會(huì)立刻失敗，讓調(diào)用者選擇策略。

Netty的策略是嘗試最多寫16次，如果不成功，則暫時(shí)停掉IO線程的寫操作，等待連接可寫時(shí)再寫，writeSpinCount默認(rèn)是16，可以通過(guò)參數(shù)調(diào)整。

for?(int?i?=?writeSpinCount;?i?>?0;?i?--)?{
????????????????????????localWrittenBytes?=?buf.transferTo(ch);
????????????????????????if?(localWrittenBytes?!=?0)?{
????????????????????????????writtenBytes?+=?localWrittenBytes;
????????????????????????????break;
????????????????????????}
????????????????????????if?(buf.finished())?{
????????????????????????????break;
????????????????????????}
????????????????????}

????????????????????if?(buf.finished())?{
????????????????????????//?Successful?write?-?proceed?to?the?next?message.
????????????????????????buf.release();
????????????????????????channel.currentWriteEvent?=?null;
????????????????????????channel.currentWriteBuffer?=?null;
????????????????????????evt?=?null;
????????????????????????buf?=?null;
????????????????????????future.setSuccess();
????????????????????}?else?{
????????????????????????//?Not?written?fully?-?perhaps?the?kernel?buffer?is?full.
????????????????????????addOpWrite?=?true;
????????????????????????channel.writeSuspended?=?true;

11.數(shù)據(jù)傳輸

數(shù)據(jù)在網(wǎng)絡(luò)上傳輸主要取決于帶寬和網(wǎng)絡(luò)環(huán)境。

12.客戶端IO線程把數(shù)據(jù)從緩沖區(qū)讀出

這個(gè)過(guò)程跟流程6是一樣的

13.IO線程把數(shù)據(jù)交給Dubbo業(yè)務(wù)線程池

這一步與流程7是一樣的，這個(gè)線程池名字為DubboClientHandler。

14.業(yè)務(wù)線程池根據(jù)消息ID通知主線程

先通過(guò)HeaderExchangeHandler的received函數(shù)得知是Response，然后調(diào)用handleResponse，

public?class?HeaderExchangeHandler?implements?ChannelHandlerDelegate?{
????static?void?handleResponse(Channel?channel,?Response?response)?throws?RemotingException?{
????????if?(response?!=?null?&&?!response.isHeartbeat())?{
????????????DefaultFuture.received(channel,?response);
????????}
????}
????public?void?received(Channel?channel,?Object?message)?throws?RemotingException?{
????????......
????????if?(message?instanceof?Response)?{
????????????????handleResponse(channel,?(Response)?message);
????????}
????????......
}

DefaultFuture根據(jù)ID獲取Future，通知調(diào)用線程

public?static?void?received(Channel?channel,?Response?response)?{
?????????......
?????????DefaultFuture?future?=?FUTURES.remove(response.getId());
?????????if?(future?!=?null)?{
????????????future.doReceived(response);
?????????}
?????????......
????}

至此，主線程獲取了返回?cái)?shù)據(jù)，調(diào)用結(jié)束。

三、影響上述流程的關(guān)鍵參數(shù)

協(xié)議參數(shù)

我們?cè)谑褂肈ubbo時(shí)，需要在服務(wù)端配置協(xié)議，例如

"dubbo"?port="20880"?dispatcher="all"?threadpool="fixed"?threads="2000"?/>

下面是協(xié)議中與性能相關(guān)的一些參數(shù)，在我們的使用場(chǎng)景中，線程池選用了fixed，大小是500，隊(duì)列為0，其他都是默認(rèn)值。

服務(wù)參數(shù)

針對(duì)每個(gè)Dubbo服務(wù)，都會(huì)有一個(gè)配置，全部的參數(shù)配置在這：http://dubbo.apache.org/zh-cn/docs/user/references/xml/dubbo-service.html。

我們關(guān)注幾個(gè)與性能相關(guān)的。在我們的使用場(chǎng)景中，重試次數(shù)設(shè)置成了0，集群方式用的failfast，其他是默認(rèn)值。

這次擁堵的主要原因，應(yīng)該就是服務(wù)的connections設(shè)置的太小，dubbo不提供全局的連接數(shù)配置，只能針對(duì)某一個(gè)交易做個(gè)性化的連接數(shù)配置。

四、連接數(shù)與Socket緩沖區(qū)對(duì)性能影響的實(shí)驗(yàn)

通過(guò)簡(jiǎn)單的Dubbo服務(wù)，驗(yàn)證一下連接數(shù)與緩沖區(qū)大小對(duì)傳輸性能的影響。

我們可以通過(guò)修改系統(tǒng)參數(shù)，調(diào)節(jié)TCP緩沖區(qū)的大小。

在 /etc/sysctl.conf 修改如下內(nèi)容， tcp_rmem是發(fā)送緩沖區(qū)，tcp_wmem是接收緩沖區(qū)，三個(gè)數(shù)值表示最小值，默認(rèn)值和最大值，我們可以都設(shè)置成一樣。

net.ipv4.tcp_rmem?=?4096?873800?16777216
net.ipv4.tcp_wmem?=?4096?873800?16777216

然后執(zhí)行sysctl –p 使之生效。

服務(wù)端代碼如下，接受一個(gè)報(bào)文，然后返回兩倍的報(bào)文長(zhǎng)度，隨機(jī)sleep 0-300ms，所以均值應(yīng)該是150ms。服務(wù)端每10s打印一次tps和響應(yīng)時(shí)間，這里的tps是指完成函數(shù)調(diào)用的tps，而不涉及傳輸，響應(yīng)時(shí)間也是這個(gè)函數(shù)的時(shí)間。

??//服務(wù)端實(shí)現(xiàn)
???public?String?sayHello(String?name)?{????
????????counter.getAndIncrement();
????????long?start?=?System.currentTimeMillis();
????????try?{
????????????Thread.sleep(rand.nextInt(300));
????????}?catch?(InterruptedException?e)?{??????
????????}
????????String?result?=?"Hello?"?+?name?+?name??+?",?response?form?provider:?"?+?RpcContext.getContext().getLocalAddress();
????????long?end?=?System.currentTimeMillis();
????????timer.getAndAdd(end-start);
????????return?result;
????}

客戶端起N個(gè)線程，每個(gè)線程不停的調(diào)用Dubbo服務(wù)，每10s打印一次qps和響應(yīng)時(shí)間，這個(gè)qps和響應(yīng)時(shí)間是包含了網(wǎng)絡(luò)傳輸時(shí)間的。

?for(int?i?=?0;?i????????????threads[i]?=?new?Thread(new?Runnable()?{
???????????????@Override
???????????????public?void?run()?{
???????????????????while(true)?{
???????????????????????Long?start?=?System.currentTimeMillis();
???????????????????????String?hello?=?service.sayHello(z);?
???????????????????????Long?end?=?System.currentTimeMillis();
???????????????????????totalTime.getAndAdd(end-start);
???????????????????????counter.getAndIncrement();
???????????????????}
???????????????}});
???????????threads[i].start();
???????}

通過(guò)ss -it命令可以看當(dāng)前tcp socket的詳細(xì)信息，包含待對(duì)端回復(fù)ack的數(shù)據(jù)Send-Q，最大窗口cwnd，rtt(round trip time)等。

(base)?niuxinli@ubuntu:~$?ss?-it
State????????????????????????????Recv-Q????????????????????????Send-Q???????????????????????????????????????????????????????Local?Address:Port??????????????????????????????????????????????????????????Peer?Address:Port?????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????
ESTAB????????????????????????????0?????????????????????????????36?????????????????????????????????????????????????????????????192.168.1.7:ssh????????????????????????????????????????????????????????????192.168.1.4:58931???????????????????????
?????cubic?wscale:8,2?rto:236?rtt:33.837/8.625?ato:40?mss:1460?pmtu:1500?rcvmss:1460?advmss:1460?cwnd:10?bytes_acked:559805?bytes_received:54694?segs_out:2754?segs_in:2971?data_segs_out:2299?data_segs_in:1398?send?3.5Mbps?pacing_rate?6.9Mbps?delivery_rate?44.8Mbps?busy:36820ms?unacked:1?rcv_rtt:513649?rcv_space:16130?rcv_ssthresh:14924?minrtt:0.112
ESTAB????????????????????????????0?????????????????????????????0??????????????????????????????????????????????????????????????192.168.1.7:36666??????????????????????????????????????????????????????????192.168.1.7:2181????????????????????????
?????cubic?wscale:7,7?rto:204?rtt:0.273/0.04?ato:40?mss:33344?pmtu:65535?rcvmss:536?advmss:65483?cwnd:10?bytes_acked:2781?bytes_received:3941?segs_out:332?segs_in:170?data_segs_out:165?data_segs_in:165?send?9771.1Mbps?lastsnd:4960?lastrcv:4960?lastack:4960?pacing_rate?19497.6Mbps?delivery_rate?7621.5Mbps?app_limited?busy:60ms?rcv_space:65535?rcv_ssthresh:66607?minrtt:0.035
ESTAB????????????????????????????0?????????????????????????????27474??????????????????????????????????????????????????????????192.168.1.7:20880??????????????????????????????????????????????????????????192.168.1.5:60760???????????????????????
?????cubic?wscale:7,7?rto:204?rtt:1.277/0.239?ato:40?mss:1448?pmtu:1500?rcvmss:1448?advmss:1448?cwnd:625?ssthresh:20?bytes_acked:96432644704?bytes_received:49286576300?segs_out:68505947?segs_in:36666870?data_segs_out:67058676?data_segs_in:35833689?send?5669.5Mbps?pacing_rate?6801.4Mbps?delivery_rate?627.4Mbps?app_limited?busy:1340536ms?rwnd_limited:400372ms(29.9%)?sndbuf_limited:433724ms(32.4%)?unacked:70?retrans:0/5?rcv_rtt:1.308?rcv_space:336692?rcv_ssthresh:2095692?notsent:6638?minrtt:0.097

通過(guò)netstat -nat也能查看當(dāng)前tcp socket的一些信息，比如Recv-Q, Send-Q。

(base)?niuxinli@ubuntu:~$?netstat?-nat
Active?Internet?connections?(servers?and?established)
Proto?Recv-Q?Send-Q?Local?Address???????????Foreign?Address?????????State??????
tcp????????0??????0?0.0.0.0:20880???????????0.0.0.0:*???????????????LISTEN?????
tcp????????0?????36?192.168.1.7:22??????????192.168.1.4:58931???????ESTABLISHED
tcp????????0??????0?192.168.1.7:36666???????192.168.1.7:2181????????ESTABLISHED
tcp????????0??65160?192.168.1.7:20880???????192.168.1.5:60760???????ESTABLISHED

可以看以下Recv-Q和Send-Q的具體含義：

Recv-Q
???????Established:?The?count?of?bytes?not?copied?by?the?user?program?connected?to?this?socket.??

???Send-Q
???????Established:?The?count?of?bytes?not?acknowledged?by?the?remote?host.

Recv-Q是已經(jīng)到了接受緩沖區(qū)，但是還沒被應(yīng)用代碼讀走的數(shù)據(jù)。Send-Q是已經(jīng)到了發(fā)送緩沖區(qū)，但是對(duì)方還沒有回復(fù)Ack的數(shù)據(jù)。這兩種數(shù)據(jù)正常一般不會(huì)堆積，如果堆積了，可能就有問(wèn)題了。

第一組實(shí)驗(yàn)：?jiǎn)芜B接，改變TCP緩沖區(qū)

結(jié)果：

繼續(xù)調(diào)大緩沖區(qū)

我們用netstat或者ss命令可以看到當(dāng)前的socket情況，下面的第二列是Send-Q大小，是寫入緩沖區(qū)還沒有被對(duì)端確認(rèn)的數(shù)據(jù)，發(fā)送緩沖區(qū)最大時(shí)64k左右，說(shuō)明緩沖區(qū)不夠用。

繼續(xù)增大緩沖區(qū)，到4M，我們可以看到，響應(yīng)時(shí)間進(jìn)一步下降，但是還是在傳輸上浪費(fèi)了不少時(shí)間，因?yàn)榉?wù)端應(yīng)用層沒有壓力。

服務(wù)端和客戶端的TCP情況如下，緩沖區(qū)都沒有滿

這個(gè)時(shí)候，再怎么調(diào)大TCP緩沖區(qū)，也是沒用的，因?yàn)槠款i不在這了，而在于連接數(shù)。因?yàn)樵贒ubbo中，一個(gè)連接會(huì)綁定到一個(gè)NioWorker線程上，讀寫都由這一個(gè)連接完成，傳輸?shù)乃俣瘸^(guò)了單個(gè)線程的讀寫能力，所以我們看到在客戶端，大量的數(shù)據(jù)擠壓在接收緩沖區(qū)，沒被讀走，這樣對(duì)端的傳輸速率也會(huì)慢下來(lái)。

第二組實(shí)驗(yàn)：多連接，固定緩沖區(qū)

服務(wù)端的純業(yè)務(wù)函數(shù)響應(yīng)時(shí)間很穩(wěn)定，在緩沖區(qū)較小的時(shí)候，調(diào)大連接數(shù)雖然能讓時(shí)間降下來(lái)，但是并不能到最優(yōu)，所以緩沖區(qū)不能設(shè)置太小，Linux一般默認(rèn)是4M，在4M的時(shí)候，4個(gè)連接基本上已經(jīng)能把響應(yīng)時(shí)間降到最低了。

結(jié)論

要想充分利用網(wǎng)絡(luò)帶寬， 緩沖區(qū)不能太小，如果太小有可能一次傳輸?shù)膱?bào)文就大于了緩沖區(qū)，嚴(yán)重影響傳輸效率。但是太大了也沒有用，還需要多個(gè)連接數(shù)才能夠充分利用CPU資源，連接數(shù)起碼要超過(guò)CPU核數(shù)。