一、前言

供應(yīng)鏈時效域歷經(jīng)近一年的發(fā)展，在預估時效方面沉淀出了一套理論和兩把利器（預估模型和路由系統(tǒng)）。以現(xiàn)貨為例，通過持續(xù)的技術(shù)方案升級，預估模型的準確率最高接近了90%，具備了透出給用戶的條件。但在接入前臺場景的過程中，前臺對我們提出接口性能的要求。

以接入的商詳浮層場景為例，接口調(diào)用鏈路經(jīng)過商詳、出價、交易，給到我們供應(yīng)鏈只有15ms的時間，在15ms內(nèi)完成所有的業(yè)務(wù)邏輯處理是一個不小的挑戰(zhàn)。

二、初始狀態(tài) - 春風得意馬蹄疾

拋開業(yè)務(wù)場景聊接口性能就是耍流氓。時效預估接口依賴于很多數(shù)據(jù)源：模型基礎(chǔ)數(shù)據(jù)、模型兜底數(shù)據(jù)、倉庫數(shù)據(jù)、SPU類目數(shù)據(jù)、賣家信息數(shù)據(jù)等，如何快速批量獲取到內(nèi)存中進行邏輯運算，是性能提升的關(guān)鍵。

最先接入時效表達的是現(xiàn)貨業(yè)務(wù)，最初的查詢單個現(xiàn)貨SKU時效的接口調(diào)用鏈路如下：

根據(jù)trace分析，接口性能的瓶頸在于數(shù)據(jù)查詢，而不在于邏輯處理，數(shù)據(jù)查詢后的邏輯處理耗時只占0.6%。

數(shù)據(jù)查詢又分為外部查詢和內(nèi)部查詢。外部查詢?yōu)?次RPC調(diào)用（耗時占比27%），內(nèi)部查詢?yōu)?1次DB查詢（耗時占比73%）。

為什么會有這么多次內(nèi)部查詢？因為預估模型是分段的，每段又根據(jù)不同的影響因子有不同的兜底策略，無法聚合成一次查詢。

單個SKU時效查詢都達到了76.5ms，以商詳浮層頁30個現(xiàn)貨SKU時效批量查詢估算，一次請求需要76.5*30=2295ms，這是不可接受的，性能提升刻不容緩。

三、優(yōu)化Round

1 - 昨夜西風凋碧樹

3.1 內(nèi)部查詢優(yōu)化

由于內(nèi)部查詢需要的預估模型數(shù)據(jù)都是離線清洗，按天級別同步的，對實時性要求不高，有多種方案可以選擇：

最終選擇方案二，離線數(shù)據(jù)同步到Redis中。由于模型數(shù)據(jù)量增幅不大，每天的同步更多的是覆蓋，故采用32G實例完全能滿足要求。

3.2 外部查詢優(yōu)化

將三個RPC查詢接口逐個分析，找到優(yōu)化方案：

3.3優(yōu)化后效果

優(yōu)化后的效果很明顯，單個SKU時效查詢RT已從76.5ms降低至27ms，同時減少了對外部域的直接依賴，一定程度上提升了穩(wěn)定性。

27ms仍然沒法滿足要求。當前的瓶頸在查詢Redis上（耗時占比96%），是否可以再進一步優(yōu)化？

四、優(yōu)化Round

2 - 衣帶漸寬終不悔

通過上述分析，可以看到目前的耗時集中在一次次的Redis I/O操作中，如果將一組Redis命令進行組裝，通過一次傳輸給Redis并返回結(jié)果，可以大大地減少耗時。

4.1 pipeline原理

Redis客戶端執(zhí)行一條命令分為如下四個過程：

1）發(fā)送命令

2）命令排隊

3）執(zhí)行命令

4）返回結(jié)果

其中1-4稱為Round Trip Time（RTT，往返時間）。pipeline通過一次性將多Redis命令發(fā)往Redis服務(wù)端，大大減少了RTT。

4.2優(yōu)化和效果

雖然Redis提供了像mget、mset這種批量接口，但Redis不支持hget批量操作，且不支持mget、hget混合批量查詢，只能采用pipeline。另外我們的場景是多key讀場景，并且允許一定比例（少概率事件）讀失敗，且pipeline中的其中一條讀失敗（pipeline是非原子性的），也不會影響時效預估，因為有兜底策略，故非常適合。

由于Redis查詢之間存在相互依賴，上次查詢的結(jié)果需要作為下次查詢的入?yún)ⅲ薀o法將所有redis查詢合并成一個Redis pipeline。雖然最終仍然存在3次Redis I/O，但7ms的RT滿足了要求。

4.3 代碼

// pipeline查詢類public class RedisBasePipelineRegister {    // 存放查到的數(shù)據(jù)    private ThreadLocal<Map<String, Object>> context = ThreadLocal.withInitial(HashMap::new);
    // 查詢    public void fetch(final RedisConsumers redisConsumers){        if (redisConsumers.isNotEmpty()){            List<Object> ret = redisClient.executePipelined((RedisCallback<Object>) connection -> {                connection.openPipeline();                redisConsumers.get().forEach(t -> t.accept(connection));                return null;            });            addValueToContext(ret,redisConsumers.getKeyList());        }    }
    /**     * 將pipeline查到的數(shù)據(jù)存入threadlocal中     * 注意，redis讀取的數(shù)據(jù)可能是空的，如果是空，會填充一個null obj，這樣可以防止后面用的時候，發(fā)現(xiàn)thread local里面沒有數(shù)據(jù)，重新查redis     */    private void addValueToContext(List<Object> val, List<String> keys) {        Map<String, Object> t = context.get();        IntStream.range(0, keys.size())                .forEach(i -> t.put(keys.get(i), val.get(i) == null ? NULL_OBJ : val.get(i)));    }
    public Object get(String key) {        return context.get().get(key);    }}

// redis查詢類public class RedisClient {
    // threadlocal沒查到，再查redis（兜底）    public Object get(String key) {        Object value = Optional.ofNullable(redisBatchPipelineRegister.get(key)).orElseGet(()->                redisTemplate.opsForValue().get(key)        );        return value;    }}

即使pipeline部分失敗后，可用Redis單指令查詢作為兜底。

五、優(yōu)化Round

3 - 眾里尋他千百度

5.1 背景

隨著時效預估的準確率在寄售、品牌直發(fā)、保稅等業(yè)務(wù)場景中滿足要求后，越來越多的業(yè)務(wù)類型需要接入時效表達接口。最初為了快速上線，交易在內(nèi)部根據(jù)出價類型串行多次調(diào)時效預估接口，導致RT壓力越來越大。出于領(lǐng)域內(nèi)聚考慮，與交易開發(fā)討論后，由時效域提供不同出價類型的聚合接口，同時保證聚合接口的RT性能。

自此，進入并發(fā)區(qū)域。

5.2 ForkJoinPool vs ThreadPoolExecutor

Java7 提供了ForkJoinPool來支持將一個任務(wù)拆分成多個“小任務(wù)”并行計算，再把多個“小任務(wù)的結(jié)果合并成總的計算結(jié)果。ForkJoinPool的工作竊取是指在每個線程中會維護一個隊列來存放需要被執(zhí)行的任務(wù)。當線程自身隊列中的任務(wù)都執(zhí)行完畢后，它會從別的線程中拿到未被執(zhí)行的任務(wù)并幫助它執(zhí)行，充分利用多核CPU的優(yōu)勢。下圖為ForkJoinPool執(zhí)行示意：

而Java8的并行流采用共享線程池（默認也為ForkJoinPool線程池），性能不可控，故不考慮。

選定ThreadPoolExecutor后，需要考慮如何設(shè)計參數(shù)。根據(jù)實際情況分析，交易請求時效QPS峰值為1000左右，而我們一個請求一般會拆分3~5個線程任務(wù)，不考慮機器數(shù)的情況下，每秒任務(wù)數(shù)量：taskNum = 3000~5000。單個任務(wù)耗時taskCost = 0.01s 。上游容忍最大響應(yīng)時間 responseTime = 0.015s。

1）核心線程數(shù) = 每秒任務(wù)數(shù) * 單個任務(wù)耗時

corePoolSize = taskNum * taskCost = (3000 ~ 5000) * 0.01 = 30 ~ 50，取40

2）任務(wù)隊列容量 = 核心線程數(shù) / 單個任務(wù)耗時 * 容忍最大響應(yīng)時間

queueCapacity = corePoolSize / taskCost * responseTime = 40 / 0.01 * 0.015 = 60

3）最大線程數(shù) = （每秒最大任務(wù)數(shù) - 任務(wù)隊列容量）* 每個任務(wù)耗時

maxPoolSize = (5000 - 60) * 0.01 ≈  50

當然上述計算都是理論值，實際有可能會出現(xiàn)未達最大線程數(shù)，cpu load就打滿的情況，需要根據(jù)壓測數(shù)據(jù)來最終確定ThreadPoolExecutor的參數(shù)。

5.3優(yōu)化和壓測

經(jīng)優(yōu)化和壓測后聚合接口平均RT從22.8ms（串行）降低為8.52ms（并行），99線為13.22ms，滿足要求。

按單機300QPS（高于預估峰值QPS兩倍左右）進行壓測，接口性能和線程池運行狀態(tài)均滿足。

最終優(yōu)化后應(yīng)用內(nèi)調(diào)用鏈路示意圖如下：

5.4 代碼

// 并行時效預估類public class ConcurrentEstimateCaller {    // 自定義線程池    private Executor executor;    // 時效預估策略工廠    private EstimateStrategyFactory estimateStrategyFactory;
    //存放異步返回的結(jié)果，KEY為出價類型，VALUE為對應(yīng)的時效結(jié)果        private ConcurrentHashMap<String, CompletableFuture<List<PromiseEstimateRes>>> futures = new ConcurrentHashMap<>();
    // 提交并行任務(wù)public ConcurrentEstimateCaller submit(PromiseEstimateAggreRequest request)        for (String scene : request.getMap().keySet()) {            futures.put(scene, CompletableFuture.supplyAsync(() -> {
                EstimateStrategy estimateStrategy = estimateStrategyFactory.getStrategy(scene);
                if (estimateStrategy != null) {                    Result<PromiseEstimateBatchResponse> tmp =                            estimateStrategy.promiseEstimateBatch(EstimateAggreConvertor.INSTANCE.convertBatchRequest(request, scene));                    if (Result.SUCCESS_CODE.equals(tmp.getCode())) {                        return tmp.getData().getEstimateRes();                    }                }                return null;
            }, executor));        }    }
    // 指定時間內(nèi)等待和獲取所有子任務(wù)返回結(jié)果    public Map<String, List<PromiseEstimateRes>> join(long timeout, TimeUnit unit) throws Exception {        // 等待所有的子任務(wù)執(zhí)行完成        CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[]{})).get(timeout, unit);
        Map<String, List<PromiseEstimateRes>> res = new HashMap<>();
        for (Map.Entry<String, CompletableFuture<List<PromiseEstimateRes>>> entry : futures.entrySet()) {
            if (entry.getValue().get() != null) {                res.put(entry.getKey(), entry.getValue().get());            }        }
        return res;    }}

六、總結(jié)

接口性能進階之路隨著業(yè)務(wù)的變化和技術(shù)的升級永無止境。

分享一些建設(shè)過程中的Tips:

如果Redis和服務(wù)機器不在同一個區(qū)域會增加幾ms的跨區(qū)傳輸耗時，所以對RT敏感的場景，如果機器不同于Redis區(qū)域，可以讓運維幫忙重建機器。

阻塞隊列可以采用SynchronousQueue來提高響應(yīng)時間，但需要保證有足夠多的消費者（線程池里的消費者），并且總是有一個消費者準備好獲取交付的工作，才適合使用。

后續(xù)建設(shè)的一些思路：隨著業(yè)務(wù)和流量的增長，線程池參數(shù)如何在不重啟機器的情況下自動調(diào)整，可以參考美團開源的DynamicTp項目對線程池動態(tài)化管理，同時添加監(jiān)控、告警等功能。