Netty高級功能（一）：流控和流量整形

這一章節(jié)秧了，我們通過例子學(xué)習(xí)netty的一些高級特性。

1序无、netty客戶端流控

在有些場景下验毡，由于各種原因，會導(dǎo)致客戶端消息發(fā)送積壓帝嗡，進而導(dǎo)致OOM晶通。

  • 1、當netty服務(wù)端并發(fā)壓力過大哟玷，超過了服務(wù)端的處理能力時狮辽，channel中的消息服務(wù)端不能及時消費，這時channel堵塞巢寡，客戶端消息就會堆積在發(fā)送隊列中
  • 2喉脖、網(wǎng)絡(luò)瓶頸，當客戶端發(fā)送速度超過網(wǎng)絡(luò)鏈路處理能力抑月，會導(dǎo)致客戶端發(fā)送隊列積壓
  • 3树叽、當對端讀取速度小于己方發(fā)送速度，導(dǎo)致自身TCP發(fā)送緩沖區(qū)滿谦絮，頻繁發(fā)生write 0字節(jié)時题诵，待發(fā)送消息會在netty發(fā)送隊列中排隊

這三種情況下须误，如果客戶端沒有流控保護，這時候就很容易發(fā)生內(nèi)存泄露仇轻。

原因：

在我們調(diào)用channel的write和writeAndFlush時
io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object, io.netty.channel.ChannelPromise)，如果發(fā)送方為業(yè)務(wù)線程奶甘，則將發(fā)送操作封裝成WriteTask（繼承Runnable）篷店，放到Netty的NioEventLoop中執(zhí)行，當NioEventLoop無法完成如此多的消息的發(fā)送的時候臭家，發(fā)送任務(wù)隊列積壓疲陕，進而導(dǎo)致內(nèi)存泄漏。

解決方案：

為了防止在高并發(fā)場景下钉赁，由于服務(wù)端處理慢導(dǎo)致的客戶端消息積壓蹄殃，客戶端需要做并發(fā)保護，防止自身發(fā)生消息積壓你踩。Netty提供了一個高低水位機制诅岩，可以實現(xiàn)客戶端精準的流控。

io.netty.channel.ChannelConfig#setWriteBufferHighWaterMark 高水位
io.netty.channel.ChannelConfig#setWriteBufferLowWaterMark 低水位

當發(fā)送隊列待發(fā)送的字節(jié)數(shù)組達到高水位時带膜，對應(yīng)的channel就變?yōu)椴豢蓪憼顟B(tài)吩谦，由于高水位并不影響業(yè)務(wù)線程調(diào)用write方法把消息加入到待發(fā)送隊列，因此在消息發(fā)送時要先對channel的狀態(tài)進行判斷（ctx.channel().isWritable）膝藕。

這里涉及到的知識點是netty的消息發(fā)送機制式廷。

netty的消息發(fā)送機制

業(yè)務(wù)調(diào)用write方法后，經(jīng)過ChannelPipeline職責(zé)鏈處理芭挽，消息被投遞到發(fā)送緩沖區(qū)待發(fā)送滑废，調(diào)用flush之后會執(zhí)行真正的發(fā)送操作，底層通過調(diào)用Java NIO的SocketChannel進行非阻塞write操作袜爪，將消息發(fā)送到網(wǎng)絡(luò)上蠕趁，

image.png

當用戶線程（業(yè)務(wù)線程）發(fā)起write操作時，Netty會進行判斷饿敲，如果發(fā)現(xiàn)不少NioEventLoop（I/O線程）妻导，則將發(fā)送消息封裝成WriteTask，放入NioEventLoop的任務(wù)隊列怀各，由NioEventLoop線程執(zhí)行倔韭，代碼如下

io.netty.channel.AbstractChannelHandlerContext#write(java.lang.Object, io.netty.channel.ChannelPromise)

    @Override
    public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
        if (msg == null) {
            throw new NullPointerException("msg");
        }

        if (isNotValidPromise(promise, true)) {
            ReferenceCountUtil.release(msg);
            // cancelled
            return promise;
        }

        write(msg, true, promise);

        return promise;
    }

 private void write(Object msg, boolean flush, ChannelPromise promise) {
        AbstractChannelHandlerContext next = findContextOutbound();
        final Object m = pipeline.touch(msg, next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            if (flush) {
                next.invokeWriteAndFlush(m, promise);
            } else {
                next.invokeWrite(m, promise);
            }
        } else {
            AbstractWriteTask task;
            if (flush) {
                task = WriteAndFlushTask.newInstance(next, m, promise);
            }  else {
                task = WriteTask.newInstance(next, m, promise);
            }
            safeExecute(executor, task, promise, m);
        }
    }
 private static void safeExecute(EventExecutor executor, Runnable runnable, ChannelPromise promise, Object msg) {
        try {
//這里的executor執(zhí)行的是netty自己實現(xiàn)的SingleThreadEventExecutor#execute方法，
            executor.execute(runnable);
        } catch (Throwable cause) {
            try {
                promise.setFailure(cause);
            } finally {
                if (msg != null) {
                    ReferenceCountUtil.release(msg);
                }
            }
        }
    }

io.netty.util.concurrent.SingleThreadEventExecutor#execute

@Override
  public void execute(Runnable task) {
      if (task == null) {
          throw new NullPointerException("task");
      }

      boolean inEventLoop = inEventLoop();
      if (inEventLoop) {
          addTask(task);
      } else {
          startThread();
          addTask(task);
          if (isShutdown() && removeTask(task)) {
              reject();
          }
      }

      if (!addTaskWakesUp && wakesUpForTask(task)) {
          wakeup(inEventLoop);
      }
  }

Netty的NioEventLoop線程內(nèi)部維護了一個Queue<Runnable> taskQuue瓢对，除了處理網(wǎng)絡(luò)IO讀寫操作寿酌，同時還負責(zé)執(zhí)行網(wǎng)絡(luò)讀寫相關(guān)的Task，NioEventLoop遍歷taskQueue硕蛹，執(zhí)行消息發(fā)送任務(wù)醇疼，代碼調(diào)用入路徑如下硕并，具體的就不貼了，太長了
io.netty.channel.nio.NioEventLoop#run
-----> io.netty.util.concurrent.SingleThreadEventExecutor#runAllTasks(long)
----->io.netty.util.concurrent.AbstractEventExecutor#safeExecute
這里safeExecute執(zhí)行的task秧荆，就是前面write寫入時包裝的AbstractWriteTask倔毙，AbstractWriteTask的run中
io.netty.channel.AbstractChannelHandlerContext.AbstractWriteTask#run

經(jīng)過一些系統(tǒng)處理操作，最終會調(diào)用io.netty.channel.ChannelOutboundBuffer#addMessage方法乙濒，將發(fā)送消息加入發(fā)送隊列（鏈表）陕赃。

我們上面寫的流程從NioSocketChannel到ChnnelOutbountBuffer，實際上在這個過程中颁股，為了對發(fā)送速度和消息積壓數(shù)進行控制么库，Netty還提供了高低水位機制，當消息隊列中積壓的待發(fā)送消息總字節(jié)數(shù)到達高水位時甘有，修改Channel的狀態(tài)為不可寫诉儒，并發(fā)送通知事件；當消息發(fā)送完成后亏掀，對低水位進行判斷忱反，如果當前積壓的待發(fā)送字節(jié)數(shù)低于低水位時，則修改channel狀態(tài)為可寫滤愕，并發(fā)送通知事件缭受，具體代碼見下
io.netty.channel.ChannelOutboundBuffer#incrementPendingOutboundBytes(long)；
io.netty.channel.ChannelOutboundBuffer#decrementPendingOutboundBytes(long)该互；

image.png

總結(jié)：在實際項目中米者，根據(jù)業(yè)務(wù)QPS規(guī)劃，客戶端處理性能宇智、網(wǎng)絡(luò)帶寬蔓搞、鏈路數(shù)、消息平均碼流大小等綜合因數(shù)随橘，設(shè)置Netty高水位（setWriteBufferHighWaterMark）值喂分，可以防止在發(fā)送隊列處于高水位時繼續(xù)發(fā)送消息，導(dǎo)致積壓更嚴重机蔗，甚至發(fā)生內(nèi)存泄漏蒲祈。在系統(tǒng)中合理利用Netty的高低水位機制做消息發(fā)送的流控，既可以保護自身萝嘁，同時又能減輕服務(wù)端的壓力梆掸，可以提升系統(tǒng)的可靠性。

那么代碼中牙言，怎么使用呢酸钦？

image.png

同時在業(yè)務(wù)發(fā)送消息時，添加socketChannel.isWritable()是否可以發(fā)送判斷

    public static boolean sendMessage(String clientId,Object message){
        if(StringUtils.isEmpty(clientId)){
            log.error(" clientId 為空咱枉，找不到客戶端卑硫！");
            return false;
        }
        SocketChannel socketChannel = FactoryMap.getChannelByDevNo(clientId);
        if(socketChannel !=null ){
            if(socketChannel.isWritable()){
                socketChannel.writeAndFlush(message);
                //更新數(shù)據(jù)庫中消息狀態(tài)
                return true;
            }else {
                log.error("channel不可寫");
                return false;
            }
        }else {
            log.error(" 客戶端未連接服務(wù)器徒恋！發(fā)送消息失敗欢伏！{}",clientId);
        }
        return false;
    }

2入挣、netty服務(wù)端 流量整形

前面講的流控（高低水位控制），主要是根據(jù)發(fā)送消息隊列積壓的大小來控制客戶端channel的寫狀態(tài)硝拧，然后用戶手動根據(jù)channel.isWritable（）來控制消息是否發(fā)送财岔，用戶可以手動控制消息不能及時發(fā)送后的處理方案（比如，過期河爹、超時）。通常用在客戶端比較多桐款。

流量整形呢咸这，是一種主動調(diào)整流量輸出速度的措施，一個典型的應(yīng)用是基于下游網(wǎng)絡(luò)節(jié)點的TPS指標控制本地流量的輸出魔眨。大多數(shù)商用系統(tǒng)都由多個網(wǎng)元或者部件組成媳维，例如參與短信互動，會涉及手機遏暴，基站侄刽，短信中心，短信網(wǎng)關(guān)朋凉，SP/CP等網(wǎng)元州丹，不同網(wǎng)元或者部件的處理性能不同，為了防止突發(fā)的業(yè)務(wù)洪峰的 導(dǎo)致下游網(wǎng)元被沖垮杂彭，有時候需要消停提供流量整形功能墓毒。

image.png

Netty流量整形的主要作用：
1、防止由于上下游網(wǎng)元性能不均衡導(dǎo)致下游網(wǎng)元被沖垮亲怠，業(yè)務(wù)流程中斷所计；
2、防止由于通信模塊接收消息過快团秽，后端業(yè)務(wù)線程處理不及時主胧，導(dǎo)致出現(xiàn)“撐死”問題。
例如习勤，之前有博客的讀者咨詢過我一個問題踪栋，他們設(shè)備向服務(wù)端不間斷的上報數(shù)據(jù)，有1G左右图毕，而服務(wù)端處理不過來這么多數(shù)據(jù)己英，這種情況下，其實就可以使用流量整形來控制接收消息速度吴旋。

原理和使用

原理：攔截channelRead和write方法损肛，計算當前需要發(fā)送的消息大小厢破，對讀取和發(fā)送閾值進行判斷，如果達到了閾值治拿，則暫停讀取和發(fā)送消息摩泪，待下一個周期繼續(xù)處理，以實現(xiàn)在某個周期內(nèi)對消息讀寫速度進行控制劫谅。

使用：將流量整形ChannelHandler添加到業(yè)務(wù)解碼器之前见坑，

image.png
注意事項：

  • 全局流量整形實例只需要創(chuàng)建一次
    GlobalChannelTrafficShapingHandler 和 GlobalTrafficShapingHandler 是全局共享的，因此實例只需要創(chuàng)建一次捏检，添加到不同的ChannelPipeline即可荞驴，不要創(chuàng)建多個實例，否則流量整形將失效贯城。

  • 流量整形參數(shù)調(diào)整不要過于頻繁

  • 消息發(fā)送保護機制
    通過流量整形可以控制發(fā)送速度熊楼，但是它的控制原理是將待發(fā)送的消息封裝成Task放入消息隊列，等待執(zhí)行時間到達后繼續(xù)發(fā)送能犯，所以如果業(yè)務(wù)發(fā)送線程不判斷channle的可以狀態(tài)鲫骗，就可能會導(dǎo)致OOM問題。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 人面猴
    序言：七十年代末踩晶，一起剝皮案震驚了整個濱河市执泰，隨后出現(xiàn)的幾起案子，更是在濱河造成了極大的恐慌渡蜻，老刑警劉巖术吝，帶你破解...
    沈念sama閱讀 222,729評論 6贊 517
  • 死咒
    序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場離奇詭異茸苇，居然都是意外死亡顿苇，警方通過查閱死者的電腦和手機，發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,226評論 3贊 399
  • 救了他兩次的神仙讓他今天三更去死
    文/潘曉璐 我一進店門税弃，熙熙樓的掌柜王于貴愁眉苦臉地迎上來纪岁，“玉大人，你說我怎么就攤上這事则果♂：玻” “怎么了？”我有些...
    開封第一講書人閱讀 169,461評論 0贊 362
  • 道士緝兇錄：失蹤的賣姜人
    文/不壞的土叔 我叫張陵西壮，是天一觀的道長遗增。 經(jīng)常有香客問我，道長款青，這世上最難降的妖魔是什么做修？ 我笑而不...
    開封第一講書人閱讀 60,135評論 1贊 300
  • ?港島之戀（遺憾婚禮）
    正文 為了忘掉前任，我火速辦了婚禮，結(jié)果婚禮上饰及，老公的妹妹穿的比我還像新娘蔗坯。我一直安慰自己，他們只是感情好燎含，可當我...
    茶點故事閱讀 69,130評論 6贊 398
  • 惡毒庶女頂嫁案：這布局不是一般人想出來的
    文/花漫 我一把揭開白布宾濒。 她就那樣靜靜地躺著，像睡著了一般屏箍。 火紅的嫁衣襯著肌膚如雪绘梦。 梳的紋絲不亂的頭發(fā)上，一...
    開封第一講書人閱讀 52,736評論 1贊 312
  • 城市分裂傳說
    那天赴魁，我揣著相機與錄音卸奉，去河邊找鬼。 笑死颖御，一個胖子當著我的面吹牛榄棵，可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播郎嫁，決...
    沈念sama閱讀 41,179評論 3贊 422
  • 雙鴛鴦連環(huán)套：你想象不到人心有多黑
    文/蒼蘭香墨 我猛地睜開眼，長吁一口氣：“原來是場噩夢啊……” “哼祈噪！你這毒婦竟也來了泽铛？” 一聲冷哼從身側(cè)響起，我...
    開封第一講書人閱讀 40,124評論 0贊 277
  • 萬榮殺人案實錄
    序言：老撾萬榮一對情侶失蹤辑鲤，失蹤者是張志新（化名）和其女友劉穎盔腔，沒想到半個月后，有當?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體月褥，經(jīng)...
    沈念sama閱讀 46,657評論 1贊 320
  • ?護林員之死
    正文 獨居荒郊野嶺守林人離奇死亡弛随，尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 38,723評論 3贊 342
  • ?白月光啟示錄
    正文 我和宋清朗相戀三年，在試婚紗的時候發(fā)現(xiàn)自己被綠了宁赤。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片舀透。...
    茶點故事閱讀 40,872評論 1贊 353
  • 活死人
    序言：一個原本活蹦亂跳的男人離奇死亡，死狀恐怖决左，靈堂內(nèi)的尸體忽然破棺而出愕够，到底是詐尸還是另有隱情，我是刑警寧澤佛猛，帶...
    沈念sama閱讀 36,533評論 5贊 351
  • ?日本核電站爆炸內(nèi)幕
    正文 年R本政府宣布惑芭，位于F島的核電站，受9級特大地震影響继找，放射性物質(zhì)發(fā)生泄漏遂跟。R本人自食惡果不足惜，卻給世界環(huán)境...
    茶點故事閱讀 42,213評論 3贊 336
  • 男人毒藥：我在死后第九天來索命
    文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望幻锁。 院中可真熱鬧凯亮，春花似錦、人聲如沸越败。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,700評論 0贊 25
  • 一樁弒父案，背后竟有這般陰謀
    文/蒼蘭香墨 我抬頭看了看天上的太陽究飞。三九已至置谦，卻和暖如春，著一層夾襖步出監(jiān)牢的瞬間亿傅，已是汗流浹背媒峡。 一陣腳步聲響...
    開封第一講書人閱讀 33,819評論 1贊 274
  • 情欲美人皮
    我被黑心中介騙來泰國打工， 沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留葵擎，地道東北人谅阿。 一個月前我還...
    沈念sama閱讀 49,304評論 3贊 379
  • 代替公主和親
    正文 我出身青樓，卻偏偏與公主長得像酬滤，于是被迫代替她去往敵國和親签餐。 傳聞我的和親對象是個殘疾皇子，可洞房花燭夜當晚...
    茶點故事閱讀 45,876評論 2贊 361