導(dǎo)讀

大報(bào)文問題，在京東物流內(nèi)較少出現(xiàn)，但每次出現(xiàn)往往是大事故，甚至導(dǎo)致上下游多個(gè)系統(tǒng)故障。大報(bào)文的背后，是不同商家業(yè)務(wù)體量不同，特別是B端業(yè)務(wù)的采購及銷售出庫單，一些頭部商家對京東系統(tǒng)支持業(yè)務(wù)復(fù)雜度及容量能力的要求越來越高。因此我們有必要把這個(gè)問題重視起來，從組織上根本上解決。

一、認(rèn)識大報(bào)文問題

大報(bào)文問題，是指不同的系統(tǒng)通過網(wǎng)絡(luò)進(jìn)行數(shù)據(jù)交互時(shí)payload size過大導(dǎo)致的系統(tǒng)可用性下降問題。

??對于大報(bào)文的產(chǎn)生方，過大的報(bào)文在序列化時(shí)消耗更多內(nèi)存和CPU，在傳輸時(shí)(JSF/MQ)可能超過中間件的大小限制導(dǎo)致傳輸失??；對于大報(bào)文的消費(fèi)方，過大的報(bào)文在反序列化時(shí)會產(chǎn)生大對象，消耗更多的內(nèi)存和CPU，容易觸發(fā)FullGC甚至OOM，而在處理過程中要遍歷的內(nèi)容更多，造成響應(yīng)變慢，如果涉及數(shù)據(jù)庫操作容易產(chǎn)生大事務(wù)、慢SQL，這些容易觸發(fā)超時(shí)，如果客戶端有重試機(jī)制，會進(jìn)一步加重大報(bào)文消費(fèi)方負(fù)載，嚴(yán)重時(shí)導(dǎo)致服務(wù)集群整體不可用。

此外，由于大報(bào)文與小報(bào)文是在一個(gè)接口上完成的，使用相同的UMP key，它會導(dǎo)致監(jiān)控失真，報(bào)警閾值無效。如果日志記錄了原始報(bào)文，也可能磁盤打滿和響應(yīng)變慢。

在京東物流技術(shù)體系內(nèi)，具體表現(xiàn)為：

JMQ/JSF對payload大小的限制都屬于防御性保護(hù)措施，目前的值是科學(xué)的，它們都已經(jīng)足夠大了。在緊急止血情況下可以調(diào)整配置參數(shù)來暫時(shí)提高payload大小限制，但長期看它會加重系統(tǒng)的風(fēng)險(xiǎn)，應(yīng)該從設(shè)計(jì)入手避免超過payload大小限制。

1.1 背景知識

1.1.1 JMQ限制

根據(jù)JMQ的官方文檔，單條消息大?。篔MQ4不要超過4M，JMQ2不要超過2M。

具體原理是發(fā)送消息時(shí)在生產(chǎn)端做主動(dòng)校驗(yàn)，如果消息大小超過閾值則拋出異常(代碼實(shí)現(xiàn)與官方文檔不一致)：

class ClusterManager {
    protected volatile int maxSize = 4194304; // 4MB
}


class MessageProducer implement Producer { // Producer接口的具體實(shí)現(xiàn)類
    ClusterManager clusterManager;


    // producer.send時(shí)做校驗(yàn)
    int checkMessages(List messages) {
        int size = 0;
        for (Message message : messages) {
            size += message.getSize() // 壓縮后的大小
        }
        if (size > this.clusterManager.getMaxSize()) {
            throw new IllegalArgumentException("the total bytes of message body must be less than " + this.clusterManager.getMaxSize());
        }
    }
}

經(jīng)與JMQ團(tuán)隊(duì)確認(rèn)，JMQ消息大小的限制，以代碼實(shí)現(xiàn)為準(zhǔn)(官方文檔不準(zhǔn)確)：

??1.1.2 JSF限制

根據(jù)JSF官方文檔，JSF可以在server和consumer端分別設(shè)置payload size，默認(rèn)都是8MB。

?? 需要注意，觸發(fā)provider報(bào)文長度限制時(shí)，JSF consumer(老版本)并不會立即失敗，而是依靠客戶端超時(shí)后才返回(感覺是JSF的缺陷)。具體原因：JSF依靠底層netty來實(shí)現(xiàn)報(bào)文長度限制，當(dāng)provider從請求報(bào)文頭里取得本次請求payload size發(fā)現(xiàn)超過限定值時(shí)，不會繼續(xù)讀取報(bào)文體，而是拋出netty定義的TooLongFrameException，而該異常的處理依賴netty的ChannelHandler.exceptionCaught方法，JSF里沒有對TooLongFrameException做處理(吃掉異常)，provider端不給consumer任何響應(yīng)(請求被扔進(jìn)黑洞)，因此造成consumer一直等待響應(yīng)直到超時(shí)，而這可能把consumer端的業(yè)務(wù)線程池拖死。

class LengthFieldBasedFrameDecoder { // 基于netty io.netty.handler.codec.LengthFieldBasedFrameDecoder的改動(dòng)
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        // 從JSF協(xié)議的報(bào)文頭里獲取本次請求的payload size，此時(shí)還沒有讀取8MB的body
        long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);
        if (frameLength > maxFrameLength) { // maxFrameLength即8MB限制
            throw new TooLongFrameException();
        }
    }
}


class ServerChannelHandler implements ChannelHandler {
    public void exceptionCaught(ChannelHandlerContext ctx, final Throwable cause) {
        if (cause instanceof IOException) {
            // ...
        } else if (cause instanceof RpcException) {
            // 這里可以看到遇到這種異常，JSF是如何給consumer端響應(yīng)的
            ResponseMessage responseMessage = new ResponseMessage(); // 給consumer的響應(yīng)
            responseMessage.getMsgHeader().setMsgType(Constants.RESPONSE_MSG);
            String causeMsg = cause.getMessage();
            String channelInfo = BaseServerHandler.getKey(ctx.channel());
            String causeMsg2 = "Remote Error Channel:" + channelInfo + " cause: " + causeMsg;
            ((RpcException) cause).setErrorMsg(causeMsg2);
            responseMessage.setException(cause); // 異常傳遞給consumer
            // socket.write回consumer
            ChannelFuture channelFuture = ctx.writeAndFlush(responseMessage);
        } else {
            // TooLongFrameException會走到這里，它的繼承關(guān)系如下：
            // TooLongFrameException -> DecoderException -> CodecException -> RuntimeException
            // 異常被吃掉了，不給consumer響應(yīng)
            logger.warn("catch " + cause.getClass().getName() + " at {} : {}",
                    NetUtils.channelToString(channel.remoteAddress(), channel.localAddress()),
                    cause.getMessage());
        }
    }
}

經(jīng)與JSF團(tuán)隊(duì)確認(rèn)，consumer端或provider端發(fā)出的消息過大(超過playload)時(shí)consumer端得不到正確的異常響應(yīng)只提示請求超時(shí)的問題，已經(jīng)在1.7.5版本修復(fù)：需要provider端升級。升級后，如果consumer端發(fā)送的消息過大，provider會立即響應(yīng)RpcException。

??此外，在JSF舊版本下，consumer使用了默認(rèn)的5秒超時(shí)，但consumer拋出超時(shí)異常總用時(shí)是48秒，這是為什么？

??這是因?yàn)閏onsumer配置的timeout不包括序列化時(shí)間，這48秒是把8MB的報(bào)文序列化的耗時(shí)：

class JSFClientTransport {
    // consumer同步調(diào)用provider
    ResponseMessage send(BaseMessage msg, int timeout) {
        MsgFuture future = doSendAsyn(msg, timeout);
        return future.get(timeout, TimeUnit.MILLISECONDS);
    }


    MsgFuture doSendAsyn(final BaseMessage msg, int timeout) {
        final MsgFuture resultFuture = new MsgFuture(getChannel(), msg.getMsgHeader(), timeout);
        Protocol protocol = ProtocolFactory.getProtocol(msg.getProtocolType(), msg.getMsgHeader().getCodecType());
        byteBuf = protocol.encode(request, byteBuf); // 發(fā)送報(bào)文前的序列化
        RequestMessage request = (RequestMessage) msg;
        request.setMsg(byteBuf);
        channel.writeAndFlush(request, channel.voidPromise()); // socket.write，異步IO
        resultFuture.setSentTime(JSFContext.systemClock.now());
    }
}


class MsgFuture implements java.util.concurrent.Future {
    final long genTime = JSFContext.systemClock.now(); // new的時(shí)候就賦值了
    volatile long sentTime;


    // 拋出超時(shí)異常邏輯
    ClientTimeoutException clientTimeoutException() {
        Date now = new Date();
        String errorMsg = "[JSF-22110]Waiting provider return response timeout . Start time: " + DateUtils.dateToMillisStr(new Date(genTime))
                + ", End time: " + DateUtils.dateToMillisStr(now)
                + ", Client elapsed: " + (sentTime - genTime) // 它包括：序列化時(shí)間，由于異步IO因此不包括socket.write時(shí)間
                + "ms, Server elapsed: " + (now.getTime() - sentTime);
        return new ClientTimeoutException(errorMsg);
    }
}

1.1.3 物流網(wǎng)關(guān)限制

物流網(wǎng)關(guān)在nginx層通過client_max_body_size做了5MB限制。這意味著，JSF限制了8MB，但通過物流網(wǎng)關(guān)對外開放成HTTP JSON API時(shí)，調(diào)用者實(shí)際的限制是5MB。

1.1.4 MySQL限制

max_allowed_packet，net_buffer_length等參數(shù)在底層控制TCP層的報(bào)文長度，京東物流體系內(nèi)該值足夠大，研發(fā)不必關(guān)注。

研發(fā)需要關(guān)注的是字段長度的定義，主要是varchar的長度。MySQL通過sql_mode參數(shù)控制字段超過長度后的行為是字段截?cái)噙€是中斷事務(wù)。對于京東物流業(yè)務(wù)執(zhí)行鏈路比較長的場景來講，同一個(gè)字段可能多處保存，例如訂單行里的skuName，就會在OFC/WMS等系統(tǒng)保存，sku_name varchar長度的不一致，特殊場景下可能造成上下游交互出現(xiàn)問題。

1.1.5 其他限制

DUCC value 的長度默認(rèn)限制為 4W 字符。

UMP Key的限制128。

JMQ的businessId長度限制100，Producer在發(fā)送是默認(rèn)超時(shí)2秒，Producer發(fā)送失敗默認(rèn)重試2次。

JMQ消費(fèi)者拋出異常會導(dǎo)致重試(進(jìn)入retry-db)，首次重試10分鐘，如果重試還不成功會越來越慢推送直至過期。過期時(shí)間：JMQ2為3天，JMQ4為30天。

JSF如果不配置consumer timeout，則使用默認(rèn)值：5秒。

Zookeeper ZNode限制長度 1MB。雖然可以通過jute.maxbuffer這個(gè)Java系統(tǒng)屬性修改，但強(qiáng)烈不建議。

原則上，所有依賴的中間件都要確認(rèn)其限制約束，提升健壯性，避免邊界條件被觸發(fā)而產(chǎn)生出乎意料的錯(cuò)誤。

1.2 產(chǎn)生原因

1.2.1 集合類字段無約束

導(dǎo)致京東物流線上事故的大報(bào)文問題中，絕大部分都屬于該類問題。而這又可以細(xì)分為兩種場景：

interface JsfAPI {
    // 場景1：批量接口，對批量的大小無限制
    void foo(List requests);    
}


class Request {
    // 場景2：對一個(gè)類內(nèi)部的集合類字段大小無限制
    // JMQ產(chǎn)生大報(bào)文，絕大部分屬于該場景
    List items;
}

當(dāng)數(shù)據(jù)量增大時(shí)，報(bào)文也會增大，造成幾MB到幾十MB的報(bào)文傳輸，系統(tǒng)為了處理這樣大數(shù)據(jù)量的報(bào)文，必然會產(chǎn)生大對象，并且這種對象會一直處于內(nèi)存中，在數(shù)據(jù)保存處理時(shí)，會造成內(nèi)存不能釋放，可能觸發(fā)頻繁FullGC，CPU使用率飆升。同時(shí)，處理集合數(shù)據(jù)，往往會有數(shù)據(jù)遍歷過程，如果無并發(fā)則時(shí)間復(fù)雜度是O(N)，大的數(shù)據(jù)集必然帶來更慢的響應(yīng)速度，而consumer端不會根據(jù)payload大小動(dòng)態(tài)設(shè)置超時(shí)時(shí)間，它可能導(dǎo)致consumer端超時(shí)，超時(shí)可能帶來多次重試，進(jìn)而加重服務(wù)端壓力。

例如：無印良品訂單sku品類過多，比如一個(gè)出庫單包含2萬個(gè)sku的極端情況。

例如：WMS出庫發(fā)貨后向ECLP回傳信息，之前都是通過一個(gè)JMQ Topic: eclp_delivery進(jìn)行回傳，一份消息包含了(訂單主檔，箱明細(xì)，包裹明細(xì))3部分信息。后來中石化場景下，一個(gè)訂單的包裹明細(xì)數(shù)量非常多，導(dǎo)致ECLP處理報(bào)文時(shí)CPU飆升，同時(shí)MQ Listener與對外服務(wù)共享CPU，導(dǎo)致接單功能可用率降低。后來，從源頭入手把一個(gè)訂單按照明細(xì)進(jìn)行分頁式拆分(之前是整單回傳，之后是按明細(xì)分頁回傳)，同時(shí)把eclp_delivery這一個(gè)topic拆分成3個(gè)topic：(訂單，箱明細(xì)，包裹明細(xì))，解決了大報(bào)文問題。

1.2.2 大字段無約束

它指的是某一個(gè)字段(不是集合大小)，由于沒加長度限制，在特定場景下傳入了遠(yuǎn)超預(yù)期大小的數(shù)據(jù)而造成的故障。

ECLP的商品主數(shù)據(jù)有個(gè)下發(fā)商品的接口，有個(gè)字段skuName，接口沒有對該字段長度進(jìn)行約束。系統(tǒng)一直平穩(wěn)運(yùn)行，直到有個(gè)商家下發(fā)了某一個(gè)商品，它的skuName達(dá)到了10KB(事后發(fā)現(xiàn)，商家是把該商品詳情頁的整個(gè)HTML通過skuName傳過來了)，插入數(shù)據(jù)庫時(shí)超過了字段長度限制varchar(200)，導(dǎo)致插入失敗，但由于沒有考慮到這種場景，返回了誤導(dǎo)的錯(cuò)誤提示。展開來看，如果ECLP為skuName定義了MySQL Text類型字段，還會有更嚴(yán)重問題：ECLP接收下商品，下發(fā)給WMS，但WMS里的skuName是varchar(200)，這個(gè)問題就只能人工處理了，甚至與商家溝通。

WMS6.0為了考慮多場景全滿足，在出庫單預(yù)留了擴(kuò)展字段，在接單時(shí)技術(shù)BP自行決定寫入哪個(gè)擴(kuò)展字段。京喜BP下發(fā)出庫單時(shí)在訂單明細(xì)維度傳入了handOverSlip(交接單，其實(shí)是團(tuán)單信息，里面有多層明細(xì)嵌套)，該字段其實(shí)是一個(gè)大JSON，單個(gè)長度10KB上下，接單環(huán)節(jié)沒問題。但組建集合單會把多個(gè)出庫單組建成一個(gè)集合單，共產(chǎn)生3000多個(gè)明細(xì)，僅handOverSlip就占30MB，造成組建集合單后下發(fā)(JSF調(diào)用)揀貨時(shí)遇到了JSF 8MB限制問題，下發(fā)失敗，單據(jù)卡在那里，現(xiàn)場生產(chǎn)無法繼續(xù)。

WMS6.0的用戶中心系統(tǒng)，為其他系統(tǒng)提供了發(fā)送咚咚通知的服務(wù)，具體實(shí)現(xiàn)是調(diào)用集團(tuán)的咚咚發(fā)送接口：xxx生產(chǎn)系統(tǒng) -> 用戶中心 -> 咚咚系統(tǒng)。鏈路上每一個(gè)環(huán)節(jié)都未對通知內(nèi)容content字段長度做限制。一次xxx生產(chǎn)系統(tǒng)調(diào)用用戶中心傳入了超8MB的content字段，觸發(fā)了咚咚系統(tǒng)的JSF底層的報(bào)文限制，最終在用戶中心產(chǎn)生了ClientTimeoutException，它導(dǎo)致用戶中心的JSF業(yè)務(wù)線程池打滿；而由于用戶中心為所有業(yè)務(wù)生產(chǎn)系統(tǒng)服務(wù)，現(xiàn)場操作會依賴它，進(jìn)而導(dǎo)致生產(chǎn)卡頓，現(xiàn)場多環(huán)節(jié)無法正常生產(chǎn)。

Amazon FBA的SP-API(Sell Partner API)，對可能出現(xiàn)風(fēng)險(xiǎn)的字段都做了長度限制，例如：

String displayableOrderComment; // maxLength: 1000
String sellerSku; // maxLength: 50
String giftMessage; // maxLength: 512
String displayableComment; // maxLength: 250

1.2.3 查詢接口返回大量數(shù)據(jù)

ECLP主數(shù)據(jù)有個(gè)接口：導(dǎo)出所有warehouse list，調(diào)用方很多，訪問頻率不高，每次響應(yīng)長度3MB。該接口在線上出現(xiàn)過多次事故(2019年)。這個(gè)接口顯然是不該存在的，但把它下線需要推動(dòng)所有的調(diào)用方改動(dòng)，這個(gè)周期很長阻力也很大。

最開始，直接查數(shù)據(jù)庫，出現(xiàn)事故后加入JimDB，再次出現(xiàn)事故后配置了JimDB的local cache，后又加入JSF限流等措施。

出現(xiàn)故障時(shí)，ECLP CPU飆升，導(dǎo)致服務(wù)超時(shí)，京東零售調(diào)用方配置的超時(shí)設(shè)置很短，這導(dǎo)致越來越多的請求打過來，加重了ECLP負(fù)擔(dān)。

1.2.4 導(dǎo)出問題

這個(gè)問題與【1.2.3 查詢接口返回大量數(shù)據(jù)】看上去類似，但有很大不同：一個(gè)同步調(diào)用，返回的數(shù)據(jù)量相對少，另一個(gè)異步執(zhí)行，返回?cái)?shù)據(jù)量巨大。

WMS6.0的報(bào)表都有導(dǎo)出的需求，例如導(dǎo)出最近3個(gè)月的明細(xì)數(shù)據(jù)。貼近商家的OFC(如ECLP)，也有類似需求，商家要求導(dǎo)出明細(xì)數(shù)據(jù)。系統(tǒng)執(zhí)行過程大致是：根據(jù)用戶指定的條件異步執(zhí)行SQL，把數(shù)據(jù)庫返回的數(shù)據(jù)集寫入Excel，并存放到blob storage(指定TTL)，用戶在規(guī)定時(shí)間(TTL)內(nèi)根據(jù)storage key去blob storage下載，完成整個(gè)導(dǎo)出過程。

這里的關(guān)鍵問題是如何查詢數(shù)據(jù)庫，而數(shù)據(jù)庫作為共享資源往往是整個(gè)系統(tǒng)的瓶頸（增加復(fù)本數(shù)量意味著成本上升），它變慢會拖垮整個(gè)系統(tǒng)。如何查詢數(shù)據(jù)庫，有8個(gè)可選項(xiàng)：

??導(dǎo)出問題的本質(zhì)，是大范圍table scan，很難設(shè)計(jì)精細(xì)的復(fù)合索引。WMS6.0最初使用的是方案1，它會產(chǎn)生深分頁limit offset問題：越往后的頁面越慢，對數(shù)據(jù)庫的壓力越大。舉例：要導(dǎo)出100萬行記錄，每頁1萬，那么到50萬記錄時(shí)，每次分頁查詢相當(dāng)于數(shù)據(jù)庫要掃描50萬+行記錄后拋棄絕大部分并返回1萬行，這還要繼續(xù)執(zhí)行50次，此外分頁組件還要額外執(zhí)行count語句以計(jì)算總行數(shù)。如果每頁是1千呢？因此，數(shù)據(jù)庫的壓力被放大了，可以簡單理解為“全表掃描”了【50 + 100(count計(jì)算)=150】次，遠(yuǎn)不如不分頁（不分頁還要解決OOM問題）。目前，WMS6.0改用了方案8，根本上解決了數(shù)據(jù)庫慢查詢問題。思路是不再盲目靜態(tài)分頁，而是根據(jù)時(shí)間條件切分成多個(gè)SQL，分別查詢，保證每個(gè)SQL返回?cái)?shù)據(jù)量不大從而避免慢SQL。例如，某個(gè)倉要導(dǎo)出最近3個(gè)月的出庫單數(shù)據(jù)，那么把這1個(gè)date range拆分(explode)成N個(gè)date range，分別執(zhí)行：

condition = DateRange(from = "2022-01-01 0000", to = "2022-04-01 0000") // 用戶指定的時(shí)間范圍：3個(gè)月
// sql = select * from ob_shipment_order where xxx and update_time between condition.from and condition.to
List chunks = explode(condition)
for (DateRange chunk : chunks) {
    // 該chunk的時(shí)間范圍已經(jīng)變成了1天，甚至是1小時(shí)，具體值是根據(jù)SQL執(zhí)行計(jì)劃估算得來的：數(shù)據(jù)量越大則拆分越細(xì)
    sql = select * from ob_shipment_order where xxx and update_time between chunk.from and chunk.to
    mysql.query(sql)
}

1.2.5 payload約束不一致產(chǎn)生的問題

鏈路上經(jīng)過不同的系統(tǒng)，不同系統(tǒng)對payload size的約束不同，也可能產(chǎn)生問題，因?yàn)闆Q定是否可以正常處理的是最小的那個(gè)，但鏈路長時(shí)相關(guān)方可能不知道，在異步場景下這個(gè)問題尤為明顯。

例如，aws的API Gateway與Lambda對payload size有不同的約束，最終用戶必須知道限制最嚴(yán)格的那一個(gè)環(huán)節(jié)。

??對于京東物流，JSF與JMQ的限制不同，理論上可能產(chǎn)生這樣的問題：JSF調(diào)用者發(fā)送8MB的請求，JSF提供者處理時(shí)采用同步轉(zhuǎn)異步機(jī)制，異步把該請求8MB發(fā)送MQ，它會導(dǎo)致MQ發(fā)送永遠(yuǎn)無法成功，而JSF的調(diào)用方卻渾然不覺。

如果通過物流網(wǎng)關(guān)對外開放，網(wǎng)關(guān)nginx限制是5MB，而JSF是8MB，設(shè)計(jì)上沒問題（fail fast），但可能造成服務(wù)方承諾與調(diào)用者感知端到端的不一致。

JSF對provider(jsf:server)和consumer可以分別設(shè)置不同的報(bào)文大小限制，理論上也可能出現(xiàn)問題，但在京東物流尚未出現(xiàn)，可不必關(guān)注。

1.2.6 其他非入口場景

它發(fā)生在系統(tǒng)執(zhí)行過程內(nèi)部。典型場景是DAO層查詢數(shù)據(jù)庫返回大結(jié)果集，Redis大key問題等。這要根據(jù)具體中間件機(jī)制來識別，例如，MyBatis支持插件來識別DAO查詢出大結(jié)果集：

public class ListResultInterceptor implements org.apache.ibatis.plugin.Interceptor {
    private static final int RESULTSET_SIZE_THRESHOLD = 10000;


    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        Object result = invocation.proceed();
        if (result != null && result instanceof List) {
            int resultSetSize = ((List) result).size();
            if (resultSetSize > RESULTSET_SIZE_THRESHOLD) {
                // 報(bào)警
            }
        }


        return result;
    }
}

2.1 主動(dòng)顯式強(qiáng)約束

即，主動(dòng)防御式自我保護(hù)，而不是依靠使用者的“自覺”：外部用戶不可信賴。

對于JSF，可以通過JSR303向API Consumer顯式傳遞約束，并且該約束可以通過框架對業(yè)務(wù)代碼無侵入地自動(dòng)執(zhí)行。對于MQ，由于生產(chǎn)者與消費(fèi)者解耦，無法直接傳遞約束，只能靠主動(dòng)監(jiān)控、人工協(xié)調(diào)。

它的前提條件，是研發(fā)有能力去主動(dòng)識別出大報(bào)文風(fēng)險(xiǎn)。

2.2 Fail Fast

如果有前端，那么前端加約束，避免大報(bào)文傳遞給后端。

對于后端，鏈?zhǔn)降纳舷掠侮P(guān)系中，上游要把好關(guān)。

這個(gè)原則并不是說下游不用關(guān)心大報(bào)文問題，恰恰相反，鏈路的每個(gè)環(huán)節(jié)都要關(guān)心，但Fail Fast可以降低整體的不必要的損耗成本，也可以緩解某個(gè)環(huán)節(jié)保護(hù)機(jī)制缺失帶來的人工介入和修數(shù)成本。

2.3 上下游對齊隱式約束

同一個(gè)業(yè)務(wù)字段在上下游傳遞時(shí)，字段長度約束要一致，否則可能會出現(xiàn)上游成功落庫下游無法落庫的情況。

2.4 大報(bào)文產(chǎn)生方負(fù)責(zé)拆分

解決大報(bào)文的根本思路是拆分報(bào)文：大 -> 小。

對應(yīng)MQ來講，應(yīng)該是Producer負(fù)責(zé)拆分大報(bào)文為小報(bào)文。

對于JSF來講，有兩種情況：

consumer產(chǎn)生的大報(bào)文：應(yīng)該provider加約束，強(qiáng)迫consumer端分頁拆分請求。參考AJAX機(jī)制

典型場景：揀貨下架調(diào)用庫存預(yù)占接口，一次性傳入1萬個(gè)sku

provider產(chǎn)生的大報(bào)文：應(yīng)該變成分頁返回結(jié)果

典型場景：一次性返回所有warehouse列表

??

需要注意的是，拆分報(bào)文，會增加生產(chǎn)方和消費(fèi)方的復(fù)雜度，尤其是消費(fèi)方：冪等，集齊，(并發(fā)和異步調(diào)用時(shí)產(chǎn)生的)亂序，業(yè)務(wù)的原子性保證等。例如，一個(gè)出庫單明細(xì)行過多時(shí)，整單預(yù)占庫存(大報(bào)文) -> 按訂單明細(xì)分頁預(yù)占(小報(bào)文)。揀貨下架按明細(xì)維度分頁調(diào)用庫存預(yù)占接口場景下，如果訂單不允許缺量：整單預(yù)占時(shí)，該訂單預(yù)占庫存的原子性(要么全成功預(yù)占，要么一個(gè)sku都不預(yù)占)是由庫存系統(tǒng)(provider)保證的；而在按訂單明細(xì)維度分頁預(yù)占時(shí)，原子性需要在揀貨系統(tǒng)(consumer)保證，即如果后面頁碼的預(yù)占失敗則需要把前面頁碼的預(yù)占釋放。這增加consumer端復(fù)雜度，但為了系統(tǒng)的性能和可用性，這是值得的。當(dāng)然，也有另外一個(gè)可選方案，仍舊讓庫存保證原子性，但庫存接口需要增加類似(currentPage, totalPages)的參數(shù)，那樣就是庫存更復(fù)雜了。無論如何，都增加了整體復(fù)雜度。

三、具體辦法

3.1 報(bào)文分頁

適用場景：MQ，以及JSF返回大報(bào)文響應(yīng)。

為了保持報(bào)文的完整性，也便于消費(fèi)方實(shí)現(xiàn)冪等、集齊等邏輯，需要在報(bào)文里額外增加分頁信息：currentPage/totalPages。

class Payload {
    List items;
    int currentPage, totalPages;
}


void sendPayload(Payload payload) {
    int currentPage = 1;
    int totalPages = payload.getItems().size() / batchSize;
    Lists.partition(payload.getItems, batchSize).forEach(subItems -> {
        Payload subPayload = new Payload(subItems)
        subPayload.setPageInfo(currentPage, totalPages)
        producer.send(subPayload)
        currentPage++;
    });
}

在極端復(fù)雜場景下，也可以考慮分拆topic，但不推薦，因?yàn)樗赡茴~外引入亂序問題。

MQ報(bào)文編解碼除了目前的JSON外，也可以考慮Protobuf等更高效格式。例如京東零售訂單快照orderver就由xml升級到了PB。

3.2 報(bào)文轉(zhuǎn)存

適用場景：MQ/JSF。

這種方案，也被稱為Claim Check Pattern。

把大的明細(xì)List，按照固定batch size轉(zhuǎn)存到JFS/OSS/JimKV/S3等外部blob storage，在報(bào)文里存放指針(blob地址)列表。

class BigPayload {
    List items;
}


class SmallPayload {
    List itemBlobKeys;
}


void sendPayload(BigPayload bigPayload) {
    SmallPayload smallPayload = new SmallPayload();
    Lists.partition(bigPayload.getItems(), batchSize).forEach(subItems -> {
        List itemBlobKeys = blogStore.putObjects(subItems)
        smallPayload.addItemBlobKeys(itemBlobKeys);
    });


    producer.send(JSON.encode(smallPayload);
}

目前上游系統(tǒng)（eclp、序列號、OMC等）、DTC、下游系統(tǒng)（各版本W(wǎng)MS）的信息傳遞使用了該辦法，共用一個(gè)JFS集群。

Side effects：1)引入額外依賴，而且消費(fèi)方被迫引入依賴 2)需要Blob存儲的TTL機(jī)制或定期清理，否則加大存儲成本 3)為消費(fèi)方帶來了不確定性，從blob拿回的數(shù)據(jù)可能超大，在反序列化和處理過程中有OOM/FullGC等風(fēng)險(xiǎn)(雖然一些json庫提供了底層的基于詞法token的Streaming Parsing API，但如果要讀取全部內(nèi)容仍然耗費(fèi)大量內(nèi)存)

3.3 報(bào)文截?cái)?/strong>

class FooRequest {
    @javax.validation.constraints.Size(min = 1, max = 200)
    private List barItems;
}


interface JsfAPI {
    // 場景1：批量接口
    void foo(@javax.validation.constraints.Size(min = 1, max = 200) List requests)


    // 場景2：請求對象里有集合字段
    void bar(FooRequest request);
}

int maxJsfRetries = 3; // JSF async下的自動(dòng)重試只能應(yīng)用層自己做了
int retried = 0;
do {
    List>> futures = new LinkedList();
    Lists.partition(voList, batchSize).forEach(subVoList -> {
        ObLocatingOrderDto dto = mapper.INSTANCE.toDTO(subVoList);
        locatingAppService.outboundOrderLocate(dto); // async JSF call
        ResponseFuture> future = RpcContext.getContext().getFuture();
        futures.add(future);
    });


    for (ResponseFuture> future : futures) {
        try {
            Result result = future.get();
        } catch (RpcException jsfException) {
            retried++;
        } catch (Throwable e) {
            // 額外的業(yè)務(wù)邏輯：與JSF并發(fā)同步調(diào)用相同的處理邏輯
        }
    }
} while (retried <= maxJsfRetries);

interface JsfAPI {
    Result getSkuInfo(String sku);
    Result> listSkuInfo(List skus);
}

Result result = jsfAPI.listSkuInfo(Lists.newArrayList("EMG1800752592"));

interface JsfAPI {
    Result getSkuInfo(String sku);
}


interface JsfBulkAPI {
    Result> listSkuInfo(List skus);
}

interface MessageListener {
    void onMessage(List messages) throws Exception;
}

interface Producer {
    void send(List messages) throws JMQException;
}

/**
 * 發(fā)送分播結(jié)果消息
 */
public void send(List checkResultDtos) {
    List messageList = Lists.newArrayList();
    for (CheckResultDto checkResultDto : checkResultDtos) {
        String messageText = JmqMessage.createReportBody(checkResultDto.getUuid(), Lists.newArrayList(checkResultDto));
        messageList.add(JmqMessage.create(topic, messageText, checkResultDto.getUuid(), checkResultDto.getWarehouseNo()));
    }
    producer.send(messageList);
}

if (logger.isInfoEnabled()) {
    log.info(JsonUtil.toJson(request); // CPU intensive and disk I/O intensive(雖然日志是順序?qū)?
}

@Slf4j
public final class PayloadSizeFilter extends AbstractFilter {
    private static final int PAYLOAD_SIZE_THRESHOLD = 4 << 20; // 4MB = 8MB(JSF限制) * 50%
    private static final int BATCH_SIZE_THRESHOLD = 1000;


    @Override
    public ResponseMessage invoke(RequestMessage requestMessage) {
        if (!RpcContext.getContext().isProviderSide()) {
            // 只在provider端檢查大報(bào)文：它才是我們要保護(hù)的對象
            return getNext().invoke(requestMessage);
        }


        // 自動(dòng)識別潛在的大報(bào)文場景：針對報(bào)文大小
        Integer payloadSize = requestMessage.getMsgHeader().getLength();
        if (payloadSize != null && payloadSize > PAYLOAD_SIZE_THRESHOLD) {
            // 這里使用最簡單的日志把潛在大報(bào)文暴露出來，各團(tuán)隊(duì)可以做更細(xì)化的機(jī)制
            // 由于logbook限制只有error level日志才能配置"關(guān)鍵字報(bào)警"，這里使用log.error
            // 如果不想自動(dòng)報(bào)警，只是人工巡檢，可以log.warn
            String methodName = requestMessage.getMethodName();
            String className = requestMessage.getClassName();
            log.error("Suspected BIG payload: {}.{}, {}>{}", className, methodName, payloadSize, PAYLOAD_SIZE_THRESHOLD);
        }


        // 自動(dòng)識別潛在的大報(bào)文場景：報(bào)文字節(jié)小，但仍會導(dǎo)致處理慢，例如 List orderNos，如果發(fā)來1萬個(gè)單號？
        // 這里只能識別出入?yún)⑹荓ist的場景，對于字段類型是List的場景無效
        Invocation invocation = requestMessage.getInvocationBody();
        Class[] argClasses = invocation.getArgClasses();
        Object[] args = invocation.getArgs();
        for (int i = 0; i < argClasses.length; i++) {
            Class argClass = argClasses[i];
            if (Collection.class.isAssignableFrom(argClass)) {
                // 入?yún)㈩愋褪荂ollection
                Collection collection = (Collection) args[i];
                if (collection.size() > BATCH_SIZE_THRESHOLD) {
                    log.error("Too BIG Collection argument: {}>{}", collection.size(), BATCH_SIZE_THRESHOLD);
                }
            }
        }


        return getNext().invoke(requestMessage);
    }
}

public interface BigPayloadTrait extends MessageListener {
    int THRESHOLD_BIG_PAYLOAD = 2 << 20; // 2MB = 4MB(JMQ限制) * 50%


    default boolean suspectedBigPayload(List messages) {
        for (Message message : messages) {
            if (message.getSize() > THRESHOLD_BIG_PAYLOAD) {
                return true;
            }
        }


        return false;
    }
}