一、前言

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

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

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

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

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

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

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

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

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

三、優(yōu)化Round

1 - 昨夜西風(fēng)凋碧樹

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

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

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

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

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

3.3優(yōu)化后效果

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

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

四、優(yōu)化Round

2 - 衣帶漸寬終不悔

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

4.1 pipeline原理

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

1）發(fā)送命令

2）命令排隊(duì)

3）執(zhí)行命令

4）返回結(jié)果

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

4.2優(yōu)化和效果

雖然Redis提供了像mget、mset這種批量接口，但Redis不支持hget批量操作，且不支持mget、hget混合批量查詢，只能采用pipeline。另外我們的場(chǎng)景是多key讀場(chǎng)景，并且允許一定比例（少概率事件）讀失敗，且pipeline中的其中一條讀失敗（pipeline是非原子性的），也不會(huì)影響時(shí)效預(yù)估，因?yàn)橛卸档撞呗裕史浅＿m合。

由于Redis查詢之間存在相互依賴，上次查詢的結(jié)果需要作為下次查詢的入?yún)ⅲ薀o法將所有redis查詢合并成一個(gè)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ù)可能是空的，如果是空，會(huì)填充一個(gè)null obj，這樣可以防止后面用的時(shí)候，發(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 背景

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

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

5.2 ForkJoinPool vs ThreadPoolExecutor

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

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

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

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

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

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

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

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

maxPoolSize = (5000 - 60) * 0.01 ≈  50

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

5.3優(yōu)化和壓測(cè)

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

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

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

5.4 代碼

// 并行時(shí)效預(yù)估類public class ConcurrentEstimateCaller {    // 自定義線程池    private Executor executor;    // 時(shí)效預(yù)估策略工廠    private EstimateStrategyFactory estimateStrategyFactory;
    //存放異步返回的結(jié)果，KEY為出價(jià)類型，VALUE為對(duì)應(yīng)的時(shí)效結(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));        }    }
    // 指定時(shí)間內(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é)

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

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

如果Redis和服務(wù)機(jī)器不在同一個(gè)區(qū)域會(huì)增加幾ms的跨區(qū)傳輸耗時(shí)，所以對(duì)RT敏感的場(chǎng)景，如果機(jī)器不同于Redis區(qū)域，可以讓運(yùn)維幫忙重建機(jī)器。

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

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