一、前言

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

以接入的商詳浮層場景為例，接口調用鏈路經過商詳、出價、交易，給到我們供應鏈只有15ms的時間，在15ms內完成所有的業務邏輯處理是一個不小的挑戰。

二、初始狀態 - 春風得意馬蹄疾

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

最先接入時效表達的是現貨業務，最初的查詢單個現貨SKU時效的接口調用鏈路如下：

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

數據查詢又分為外部查詢和內部查詢。外部查詢為3次RPC調用（耗時占比27%），內部查詢為11次DB查詢（耗時占比73%）。

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

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

三、優化Round

1 - 昨夜西風凋碧樹

3.1 內部查詢優化

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

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

3.2 外部查詢優化

將三個RPC查詢接口逐個分析，找到優化方案：

3.3優化后效果

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

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

四、優化Round

2 - 衣帶漸寬終不悔

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

4.1 pipeline原理

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

1）發送命令

2）命令排隊

3）執行命令

4）返回結果

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

4.2優化和效果

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

由于Redis查詢之間存在相互依賴，上次查詢的結果需要作為下次查詢的入參，故無法將所有redis查詢合并成一個Redis pipeline。雖然最終仍然存在3次Redis I/O，但7ms的RT滿足了要求。

4.3 代碼

// pipeline查詢類public class RedisBasePipelineRegister {    // 存放查到的數據    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查到的數據存入threadlocal中     * 注意，redis讀取的數據可能是空的，如果是空，會填充一個null obj，這樣可以防止后面用的時候，發現thread local里面沒有數據，重新查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單指令查詢作為兜底。

五、優化Round

3 - 眾里尋他千百度

5.1 背景

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

自此，進入并發區域。

5.2 ForkJoinPool vs ThreadPoolExecutor

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

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

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

1）核心線程數 = 每秒任務數 * 單個任務耗時

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

2）任務隊列容量 = 核心線程數 / 單個任務耗時 * 容忍最大響應時間

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

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

maxPoolSize = (5000 - 60) * 0.01 ≈  50

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

5.3優化和壓測

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

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

最終優化后應用內調用鏈路示意圖如下：

5.4 代碼

// 并行時效預估類public class ConcurrentEstimateCaller {    // 自定義線程池    private Executor executor;    // 時效預估策略工廠    private EstimateStrategyFactory estimateStrategyFactory;
    //存放異步返回的結果，KEY為出價類型，VALUE為對應的時效結果        private ConcurrentHashMap<String, CompletableFuture<List<PromiseEstimateRes>>> futures = new ConcurrentHashMap<>();
    // 提交并行任務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));        }    }
    // 指定時間內等待和獲取所有子任務返回結果    public Map<String, List<PromiseEstimateRes>> join(long timeout, TimeUnit unit) throws Exception {        // 等待所有的子任務執行完成        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;    }}

六、總結

接口性能進階之路隨著業務的變化和技術的升級永無止境。

分享一些建設過程中的Tips:

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

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

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