1. 背景

公司有一個推薦系統Rec，這個系統的主要功能是：

向外部系統提供推薦介面
根據請求獲取推薦策略
根據推薦策略完成推薦的召回、過濾、打分、排序階段

Rec作為微服務中的一環，本身不儲存召回的物料資訊，也不儲存使用者和物料的特徵資訊，它負責就是對各個服務的組合和流轉

其流程如下：

2. 問題

在開發Rec的過程中，發現流程中存在可以優化的地方，例如：

流程	問題
「合併節點」需要等待所有的召回結果完成之後merge到一個List，然後去獲取詳情資訊	獲取詳情資訊是一個需要分批的IO操作，既然需要分批，為什麼不是每個召回完成就去獲取詳情，而要等待合併。即使不需要分批，獲取詳情也會隨著召回結果的數量變多而耗時變長
「過濾節點」需要先準備過濾的資料，等所有資料準備好，再進行過濾	獲取資料是IO操作，過濾是CPU密集操作，是不是可以獲取完一部分資料就可以立即進行過濾

上述的問題是純粹從效能方面去考慮，目前的流程從邏輯是更容易理解的

2.1 優化流程圖

合併節點

flowchart LR 畫像召回 --> 合併類目召回 --> 合併 ItemCF召回 --> 合併合併 --> 獲取詳情

flowchart LR 畫像召回 --> 獲取詳情1 類目召回 --> 獲取詳情2 ItemCF召回 --> 獲取詳情3 獲取詳情1 --> 合併獲取詳情2 --> 合併獲取詳情3 --> 合併

過濾節點

flowchart LR 獲取質量分資料 --> 完成獲取重複展示資料 --> 完成獲取黑名單資料 --> 完成完成 --> 過濾

flowchart LR 獲取質量分資料 --> 過濾1 --> 完成獲取重複展示資料 --> 過濾2 --> 完成獲取黑名單資料 --> 過濾3 --> 完成

3. 分析

我們分析上述問題，並提出了的初步的優化方式

我們還要考慮下面幾個問題：

這樣的優化是不是有效的，收益如何，從效能和程式碼理解上
如何具體實現

上面提出的兩個問題本質上是屬於流程統籌方面的問題，在這個流程中，我們還是可以找到一些別的優化的地方：

例如「路由節點獲取準備資料」是不是可以不用等待，提交完非同步任務之後直接繼續，在使用的時候去直接拿，如果發現還沒有拿到資料就等著，有資料就直接使用，而不是原來的等待拿到資料再繼續下一步

flowchart LR 畫像召回 --> id{獲取畫像資料} -->|已獲取到| 繼續 id{獲取畫像資料} -->|沒獲取到| 等待等待 --> id{獲取畫像資料}

當存在多個獲取任務時，會縮短執行的時間，舉個例子：

原先的流程：都使用執行緒池都來完成，合併前花費10ms，合併後花費10ms，總耗時20ms

flowchart LR id["獲取畫像資料完成(10ms)"] -->合併 --> id2["畫像召回(5ms)"] id1["獲取熱門類目資料完成(5ms)"]-->合併 --> id3["熱門類目召回(10ms)"]

新的流程：都使用執行緒池都來完成，總耗時15ms

flowchart LR id["獲取畫像資料完成(10ms)"] --> id2["畫像召回(5ms)"] -->合併 id1["獲取熱門類目資料完成(5ms)"] --> id3["熱門類目召回(10ms)"]-->合併

上述的流程圖展示出來和上面的提出的兩個問題屬於一類，但這個問題不能簡單改變流程，因為：

從流程圖中可以看出來它們離得太遠，而移到一起這個邏輯會被打散，沒有現在的直觀
準備資料是在準備了很多資料，但把其中幾個拿出來的效果不能確定

所以我們需要使用一些非同步程式設計的手段，在流程不變的情況下，還能使其執行的更快，下面表示優化的時間是從哪來的：

4. 調研

在調研的過程中，發現了很多相關的技術：

Reactor，Java響應式程式設計
CompletableFuture，Java非同步程式設計

Quasar，java協程

4.1 Future

Java中提供Future可以滿足我們非同步的需要嗎？

如果只是一個非同步任務，例如畫像召回需要等待畫像資料，我們可以在畫像召回中使用future.get()

但在畫像召回完成之後，我們要進行過濾，需要提前準備一些資料，例如已展示推薦資料，使用future.get()讓畫像召回變成了同步任務，在獲取結果之前無法繼續

專案目前就是使用了執行緒池+Future的方案，不過future.get()都線上程池submit之後

4.2 CompletableFuture

針對上述問題， JDK8設計出CompletableFuture。CompletableFuture提供了一種觀察者模式類似的機制，可以讓任務執行完成後通知監聽的一方

利用CompletableFuture，我們可以這樣寫：

    CompletableFuture recallFuture = CompletableFuture
                                        .supplyAsync(獲取畫像資料任務)
                                        .thenApply(畫像召回)
                                        .thenAccept(獲取召回資訊過濾資料)
    
    CompletableFuture filterDataFuture = CompletableFuture
                                        .supplyAsync(獲取過濾資料任務)
                                        
                                        
    recallFuture.thenCombineAsync(filterDataFuture, (s, w) -> {過濾資料 });

4.3 Reactor

看起來CompletableFuture已經可以滿足我們的需求了，為什麼需要再瞭解Reactor呢？它們有什麼差別？

官方參考手冊通過對比Callback、CompletableFuture和Reactor，它們都可以實現非同步功能，但CompletableFuture/Future有下面的缺點

Future objects are a bit better than callbacks, but they still do not do well at composition, despite the improvements brought in Java 8 by CompletableFuture. Orchestrating multiple Future objects together is doable but not easy. Also, Future has other problems:

It is easy to end up with another blocking situation with Future objects by calling the get() method.

They do not support lazy computation.

They lack support for multiple values and advanced error handling.

呼叫future.get()就進入到了阻塞，這種情況很容易出現
不支援惰性計算，參考StackOverFlow中Oleh Dokuka的回答
不支援多個值處理和高階錯誤處理，檢視官方參考手冊中的例子可以看出

除此之外，官方參考手冊介紹了一些其他特點，參考3.3

4.3.1 實現

如何使用Reactor實現上述功能呢？

         Mono<String> recallMono = Mono
                .fromCallable(() -> "獲取畫像資料")
                .flatMap((portraitData) -> Mono.fromCallable(() -> "畫像召回"))
                .flatMap((recItemData) -> Mono.fromCallable(() -> "獲取召回資訊過濾資料"));

        Mono<String> filterDataMono = Mono
                .fromCallable(() -> "獲取過濾資料任務");

        Mono.zip(recallMono, filterDataMono).filter((t)->true);

Java大部分的library都是同步的（HttpClient，JDBC），Mono可以和Future組合使用執行緒來實現非同步任務，Java也存在一些非同步庫例如Netty，Redis Luttuce.

4.4 Quasar

至於為什麼會提到Quasar，貝殼技術 | 響應式程式設計和協程在 Java 語言的應用中介紹了響應式程式設計和協程一起使用的場景，給出了原因：

響應式程式設計必須使用非同步才能發揮其作用
Java中非同步的唯一解決方案就是執行緒
過多的執行緒會造成OOM，所以需要使用協程

我個人覺得響應式程式設計本質是從統籌學來優化程式的，最著名的例子就是燒水泡茶流程，我們只不過是通過合理編排讓硬體資源最大化利用。

具體實現是將原本同步邏輯中的片段打散到不同的執行緒中去非同步執行，原本同步阻塞的執行緒這時候可以給別的任務使用，應該會減少更多執行緒的使用

5. 實現

使用執行緒池來處理Reactor中的非同步任務
使用flatMap、map編排後續任務
使用Flux表示推薦結果流，通過不同召回不斷把召回結果sink到流中
使用buffer來處理分批任務
使用zip或者flatMap來處理並行任務
使用distinct去重
使用block非同步轉同步獲取推薦結果

6. 效果

待更新

參考

[1] 貝殼技術 | 響應式程式設計和協程在 Java 語言的應用
[2] 非同步程式設計利器：CompletableFuture詳解｜Java 開發實戰
[3] Reactor Java檔案
[4] 並行模型之Actor和CSP
[5] RxJava VS Reactor
[6] CompletableFuture原理與實踐-外賣商家端API的非同步化

使用Spring Reactor優化推薦流程