RxJava 沉思錄（三）：時間維度

本文是 "RxJava 沉思錄" 系列的第三篇分享。本系列全部分享：

• RxJava 沉思錄（一）：你認爲 RxJava 真的好用嗎？
• RxJava 沉思錄（二）：空間維度
• RxJava 沉思錄（三）：時間維度
• RxJava 沉思錄（四）：總結

在上一篇分享中，咱們應該已經對 Observable 在空間維度上從新組織事件的能力 印象深入了，那麼天然而然的，咱們容易聯想到時間維度，事實上就我我的而言，我認爲 Observable 在時間維度上的從新組織事件的能力 相比較其空間維度的能力更爲突出。與上一篇相似，本文接下來將經過列舉真實的例子來闡述這一論點。

點擊事件防抖動

這是一個比較常見的情景，用戶在手機比較卡頓的時候，點擊某個按鈕，正常應該啓動一個頁面，可是手機比較卡，沒有當即啓動，用戶就點了好幾下，結果等手機回過神來的時候，就會啓動好幾個同樣的頁面。

這個需求用 Callback 的方式比較難處理，可是相信用過 RxJava 的開發者都知道怎麼處理：

RxView.clicks(btn)
    .debounce(500, TimeUnit.MILLISECONDS)
    .observerOn(AndroidSchedulers.mainThread())
    .subscribe(o -> {
        // handle clicks
    })
複製代碼

debounce 操做符產生一個新的 Observable, 這個 Observable 只發射原 Observable 中時間間隔小於指定閾值的最大子序列的最後一個元素。 參考資料：Debounce

雖然這個例子比較簡單，可是它很好的表達了 Observable 能夠在時間維度上對其發射的事件進行從新組織 , 從而作到以前 Callback 形式不容易作到的事情。

社交軟件上消息的點贊與取消點贊

點贊與取消點贊是社交軟件上常常出現的需求，假設咱們目前有下面這樣的點贊與取消點讚的代碼：

boolean like;

likeBtn.setOnClickListener(v -> {
    if (like) {
        // 取消點贊
        sendCancelLikeRequest(postId);
    } else {
        // 點贊
        sendLikeRequest(postId);
    }
    like = !like;
});
複製代碼

如下圖片素材資源來自 Dribbble

若是你碰巧實現了一個很是酷炫的點贊動畫，用戶可能會玩得不亦樂乎，這個時候可能會對後端服務器形成必定的壓力，由於每次點贊與取消點贊都會發起網絡請求，假如不少用戶同時在玩這個點贊動畫，服務器可能會不堪重負。

和前一個例子的防抖動思路差很少，咱們首先想到須要防抖動：

boolean like;
PublishSubject<Boolean> likeAction = PublishSubject.create();

likeBtn.setOnClickListener(v -> {
    likeAction.onNext(like);
    like = !like;
});

likeAction.debounce(1000, TimeUnit.MILLISECONDS)
    .observerOn(AndroidSchedulers.mainThread())
    .subscribe(like -> {
        if (like) {
            sendCancelLikeRequest(postId);
        } else {
            sendLikeRequest(postId);
        }
    });
複製代碼

寫到這個份上，其實已經能夠解決服務器壓力過大的問題了，可是仍是有優化空間，假設當前是已贊狀態，用戶快速點擊 2 下，按照上面的代碼，仍是會發送一次點讚的請求，因爲當前是已贊狀態，再發送一次點贊請求是沒有意義的，因此咱們優化的目標就是將這一類事件過濾掉：

Observable<Boolean> debounced = likeAction.debounce(1000, TimeUnit.MILLISECONDS);
debounced.zipWith(
    debounced.startWith(like),
    (last, current) -> last == current ? new Pair<>(false, false) : new Pair<>(true, current)
)
    .flatMap(pair -> pair.first ? Observable.just(pair.second) : Observable.empty())
    .subscribe(like -> {
        if (like) {
            sendCancelLikeRequest(postId);
        } else {
            sendLikeRequest(postId);
        }
    });
複製代碼

zipWith 操做符能夠把兩個 Observable 發射的相同序號(同爲第 x 個)的元素，進行運算轉換，獲得的新元素做爲新的 Observable 對應序號所發射的元素。參考資料：ZipWith

上面的代碼，咱們能夠看到，首先咱們對事件流作了一次 debounce 操做，獲得 debounced 事件流，而後咱們把 debounced 事件流和 debounced.startWith(like) 事件流作了一次 zipWith 操做。至關於新的這個 Observable 中發射的第 n 個元素（n >= 2）是由 debounced 事件流中的第 n 和 第 n-1 個元素運算獲得的（新的這個 Observable 中發射的第 1 個元素是由 debounced 事件流中的第 1 個元素和原始點贊狀態 like 運算而來）。

運算的結果是獲得一個 Pair 對象，它是一個雙布爾類型二元組，二元組第一個元素爲 true 表明這個事件不應被忽略，應該被觀察者觀察到；若爲 false 則應該被忽略。二元組的第二個元素僅在第一個元素爲 true 的狀況下才有意義，true 表示應該發起一次點贊操做，而 false 表示應該發起一次取消點贊操做。上面提到的「運算」具體運算的規則是，比較兩個元素，若相等，則把二元組的第一個元素置爲 false，若不相等，則把二元組的第一個元素置爲 true, 同時把二元組的第二個元素置爲 debounced 事件流發射的那個元素。

隨後的 flatMap 操做符完成了兩個邏輯，一是過濾掉二元組第一個元素爲 false 的二元組，二是把二元組轉化回最初的 Boolean 事件流。其實這個邏輯也可由 filter 和 map 兩個操做符配合完成，這裏爲了簡單用了一個操做符。

雖然上面用了很多篇幅解釋了每一個操做符的意義，但其實核心思想是簡單的，就是在原先 debounce 操做符的基礎上，把獲得的事件流裏每一個元素和它的上一個元素作比較，若是這個元素和上個元素相同（例如在已贊狀態下再次發起點贊操做）, 就把這個元素過濾掉，這樣最終的觀察者裏只會在在真正須要改變點贊狀態的時候纔會發起網絡請求了。

咱們考慮用 Callback 實現相同邏輯，雖然比較本次操做與上次操做這樣的邏輯經過 Callback 也能夠作到，可是 debounce 這個操做符完成的任務，若是要使用 Callback 來實現就很是複雜了，咱們須要定義一個計時器，還要負責啓動與關閉這個計時器，咱們的 Callback 內部會摻雜進不少和觀察者自己無關的邏輯，相比 RxJava 版本的純粹相去甚遠。

檢測雙擊事件

首先，咱們須要定義雙擊事件，不妨先規定兩次點擊小於 500 毫秒則爲一次雙擊事件。咱們先使用 Callback 的方式實現：

long lastClickTimeStamp;

btn.setOnClickListener(v -> {
    if (System.currentTimeMillis() - lastClickTimeStamp < 500) {
        // handle double click
    }
});
複製代碼

上面的代碼很容易理解，咱們引入一箇中間變量 lastClickTimeStamp, 經過比較點擊事件發生時和上一次點擊事件的時間差是否小於 500 毫秒，來確認是否發生了一次雙擊事件。那麼如何經過 RxJava 來實現呢？就和上一個例子同樣，咱們能夠在時間維度對 Observable 發射的事件進行從新組織，只過濾出與上次點擊事件間隔小於 500 毫秒的點擊事件，代碼以下：

Observable<Long> clicks = RxView.clicks(btn)
    .map(o -> System.currentTimeMillis())
    .share();
    
clicks.zipWith(clicks.skip(1), (t1, t2) -> t2 - t1)
    .filter(interval -> interval < 500)
    .subscribe(o -> {
        // handle double click
    });
複製代碼

咱們再一次用到了 zipWith 操做符來對事件流自身相鄰的兩個元素作比較，另外此次代碼中使用了 share 操做符，用來保證點擊事件的 Observable 被轉爲 Hot Observable。

在RxJava中，Observable能夠被分爲Hot Observable與Cold Observable,引用《Learning Reactive Programming with Java 8》中一個形象的比喻（翻譯後的意思）：咱們能夠這樣認爲，Cold Observable在每次被訂閱的時候爲每個Subscriber單獨發送可供使用的全部元素，而Hot Observable始終處於運行狀態當中，在它運行的過程當中，向它的訂閱者發射元素（發送廣播、事件），咱們能夠把Hot Observable比喻成一個電臺，聽衆從某個時刻收聽這個電臺開始就能夠聽到此時播放的節目以及以後的節目，可是沒法聽到電臺此前播放的節目，而Cold Observable就像音樂 CD ，人們購買 CD 的時間可能先後有差距，可是收聽 CD 時都是從第一個曲目開始播放的。也就是說同一張 CD ，每一個人收聽到的內容都是同樣的， 不管收聽時間早或晚。

僅僅是上面這個雙擊檢測的例子，還不能體現 RxJava 的優越性，咱們把需求改得更復雜一點：若是用戶在「短期」內連續屢次點擊，只能算一次雙擊操做。這個需求是合理的，由於若是按照上面 Callback 的寫法，雖然能夠檢測出雙擊操做，可是若是用戶快速點擊 n 次（間隔均小於 500 毫秒，n >= 2）, 就會觸發 n - 1 次雙擊事件，假設雙擊處理函數裏須要發起網絡請求，會對服務器形成壓力。要實現這個需求其實也簡單，和上一個例子相似，咱們用到了 debounce 操做符：

Observable<Object> clicks = RxView.clicks(btn).share()

clicks.buffer(clicks.debounce(500, TimeUnit.MILLISECONDS))
    .filter(events -> events.size >= 2)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(o -> {
        // handle double click
    });
複製代碼

buffer 操做符接受一個 Observable 爲參數，這個 Observable 所發射的元素是什麼不重要，重要的是這些元素髮射的時間點，這些時間點會在時間維度上把原來那個 Observable 所發射的元素劃分爲一系列元素的組，buffer 操做符返回的新的 Observable 發射的元素即爲那些「組」。
參考資料: Buffer

上面的代碼經過 buffer 和 debounce 兩個操做符很巧妙的把點擊事件流轉化爲了咱們關心的 「短期內點擊次數超過 2 次」 的事件流，並且新的事件流中任意兩個相鄰事件間隔一定大於 500 毫秒。

在這個例子中，若是咱們想要使用 Callback 去實現類似邏輯，代碼量確定是巨大的，並且魯棒性也沒法保證。

搜索提示

咱們平時使用的搜索框中，經常是當用戶輸入一部份內容後，下方就會顯示對應的搜索提示，以支付寶爲例，當在搜索框輸入「螞蟻」關鍵詞後，下方自動刷新和關鍵詞相關的結果：

爲了簡化這個例子，咱們不妨定義根據關鍵詞搜索的接口以下：

public interface Api {
    @GET("path/to/api")
    Observable<List<String>> queryKeyword(String keyword);
}
複製代碼

查詢接口如今已經肯定下來，咱們考慮一下在實現這個需求的過程當中須要考慮哪些因素：

1. 防止用戶輸入過快，觸發過多網絡請求，須要對輸入事件作一下防抖動。
2. 用戶在輸入關鍵詞過程當中可能觸發屢次請求，那麼，若是後一次請求的結果先返回，前一次請求的結果後返回，這種狀況應該保證界面展現的是後一次請求的結果。
3. 用戶在輸入關鍵詞過程當中可能觸發屢次請求，那麼，若是後一次請求的結果返回時，前一次請求的結果還沒有返回的狀況下，就應該取消前一次請求。

綜合考慮上面的因素之後，咱們使用 RxJava 實現的對應的代碼以下：

RxTextView.textChanges(input)
    .debounce(300, TimeUnit.MILLISECONDS)
    .switchMap(text -> api.queryKeyword(text.toString()))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> {
        // handle results
    });
複製代碼

switchMap 這個操做符與 flatMap 操做符相似，可是區別是若是原 Observable 中的兩個元素，經過 switchMap 操做符都轉爲 Observable 以後，若是後一個元素對應的 Observable 發射元素時，前一個元素對應的 Observable 還沒有發射完全部元素，那麼前一個元素對應的 Observable 會被自動取消訂閱，還沒有發射完的元素也不會體如今 switchMap 操做符調用後產生的新的 Observable 發射的元素中。 參考資料：SwitchMap

咱們分析上面的代碼，能夠發現: debounce 操做符解決了問題 1，switchMap 操做符解決了問題 2、3。這個例子能夠很好的說明，RxJava 的 Observable 能夠經過一系列操做符從時間的維度上從新組織事件，從而簡化觀察者的邏輯。這個例子若是使用 Callback 來實現，確定是十分複雜的，須要設置計時器以及一堆中間變量，觀察者中也會摻雜進不少額外的邏輯，用來保證事件與事件的依賴關係。

（未完待續）

本文屬於 "RxJava 沉思錄" 系列，歡迎閱讀本系列的其餘分享：

• RxJava 沉思錄（一）：你認爲 RxJava 真的好用嗎？
• RxJava 沉思錄（二）：空間維度
• RxJava 沉思錄（三）：時間維度
• RxJava 沉思錄（四）：總結

---

若是您對個人技術分享感興趣，歡迎關注個人我的公衆號：麻瓜日記，不按期更新原創技術分享，謝謝！:)