在 Android 上使用協程（三）：Real Work

時間 2019-11-06

原文原文鏈接

這裏是關於在 Android 上使用協程的一系列文章。本篇文章將着重於介紹使用協程來解決實際問題。css

該系列其餘文章：html

在 Android 上使用協程（一）：Getting The Backgroundandroid

在 Android 上使用協程（二）：Getting startedgit

使用協程解決現實問題

系列前兩篇文章着重於介紹協程如何簡化代碼，在 Android 上提供主線程安全，避免泄露任務。以此爲背景，對於在 Android 中處理後臺任務和簡化回調代碼，這都是一個很好的解決方案。github

到目前爲止，咱們瞭解了什麼是協程以及如何管理它們。在這篇文章中，咱們將看一下如何使用協程來完成真實的任務。協程是一種通用的編程語言特性，和函數同一級別，因此你可使用協程來實現任何對象或函數能夠完成的工做。然而，對下面這兩種真實代碼中常常出現的任務來講，協程是一個很好的解決方案。數據庫

一次性請求： 調用一次執行一次，它們老是在結果準備好以後才結束執行。
流式請求： 觀察變化並反饋給調用者，它們直到第一個結果返回纔會結束執行。

協程很好的解決了上面這些任務。這篇文章中，咱們會深刻一次性請求，探討在 Android 上如何實現它。編程

一次性請求

一次性請求每調用一次就會執行一次，結果一旦準備好就會結束執行。這和普通函數調用是同樣的模式 —— 調用，作一些工做，返回。因爲其和函數調用的類似性，它比流式請求更容易理解。後端

一次性請求每次調用時執行，結果一旦準備好就會中止執行。設計模式

舉個一次性請求的例子，想象一下你的瀏覽器是如何加載網頁的。當你點擊連接時，向服務器發送了一個網絡請求來加載網頁。一旦數據傳輸到了你的瀏覽器，它就中止與後端的交互了，此時它已經擁有了須要的全部數據。若是服務器修改了數據，新的修改不會在瀏覽器展現，你必須刷新頁面。瀏覽器

因此，即便一次性請求缺乏流式請求的實時推送功能，但它仍然很強大。在 Android 上，你可使用一次性請求作不少事情，例如查詢，存儲或者更新數據。對於列表排序來講，它也是一種好方案。

問題：展現有序列表

讓咱們經過展現有序列表來探索一次性請求。爲了使例子更加具體，咱們編寫一個產品庫存應用給商店的員工使用。它被用於根據最後一次進貨的時間來查詢貨物。貨物既能夠升序排列，也能夠降序排列。這兒的貨物太多了以致於排序花費了幾乎一秒，讓咱們使用協程來避免阻塞主線程。

這個 App 中的全部產品都存儲在數據庫 Room 中。這是一個很好的例子，由於咱們不須要進行網絡請求，這樣咱們就能夠專一於設計模式。因爲無需網絡請求使得這個例子很簡單，儘管這樣，可是它仍然展現了實現一次性請求所使用的模式。

爲了使用協程實現這個請求，你須要把協程引入 ViewModel 、Repository 和 Dao。讓咱們逐個看看它們是如何與協程結合在一塊兒的。

class ProductsViewModel(val productsRepository: ProductsRepository) : ViewModel() {
    private val _sortedProducts = MultableLiveData<List<ProductListing>>()
    val sortedProducts: LiveData<List<ProductListing>> = _sortedProducts
    
    /** * Called by the UI when the user clicks the appropriate sort button */
    fun onSortAscending() = sortPricesBy(ascending = true)
    fun onSortDescending() = sortPricesBy(ascending = false)
    
    private fun sortPricesBy(ascending: Boolean) {
        viewModelScope.launch {
            // suspend and resume make this database request main-safe
            // so our ViewModel doesn't need to worry about threading
            _sortedProducts.value = 
                productsRepository.loadSortedProducts(ascending)
        }
    }
}
複製代碼

ProductsViewModel 負責接收用戶層事件，而後請求 repository 更新數據。它使用 LiveData 存儲要在 UI 中進行展現的當前有序列表。當接收到一個新的事件，sortPricesBy 方法會開啓一個新的協程來排序集合，當結果可用時更新 LiveData。因爲 ViewModel 能夠在 onCleared 回調中取消協程，因此它是這個架構中啓動協程的好位置。當用戶離開界面的時候，就無需再繼續未完成的任務了。

若是你不是很瞭解 LiveData，這裏有一篇介紹 LiveData 如何爲 UI 層存儲數據的好文章，做者是 CeruleanOtter。

ViewModels: A Simple Example

這是在 Android 上使用協程的通用模式。因爲 Android Framework 沒法調用 suspend 函數，你須要配合一個協程來響應 UI 事件。最簡單的方法就是當事件發生時啓動一個新的協程，最適合的地方就是 ViewModel 了。

在 ViewModel 中啓動協程是一個通用的設計模式。

ViewModel 實際上經過 ProductsRepository 來獲取數據。讓咱們來看一下代碼：

class ProductsRepository(val productsDao: ProductsDao) {
    /** * This is a "regular" suspending function, which means the caller must * be in a coroutine. The repository is not responsible for starting or * stoppong coroutines since it doesn't have a natural lifecycle to cancel * unnecssary work. * * This *may* be called from Dispatchers.Main abd is main-safe because * Room will take care of main-safety for us. */
    suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
        return if (ascending) {
            productsDao.loadProductsByDateStockedAscending()
        } else {
            productsDao.loadProductsByDateStockedDescending()

        }
    }
}
複製代碼

ProductsRepository 爲商品數據的交互提供了合理的接口。在這個 App 中，因爲全部數據都是存儲在 Room 數據庫中，它提供了具備兩個針對不一樣排序的方法的 Dao。

Repository 是 Android Architecture Components 架構的可選部分。若是你在 app 中使用了 repository 或者類似做用的層級，它更偏向於使用掛起函數。因爲 repository 沒有生命週期，它僅僅只是一個對象，全部它沒有辦法作資源清理工做。在 repository 中啓動的協程將有可能泄露。

使用掛起函數，除了避免泄露之外，在不一樣上下文中也能夠重複使用 repository 。任何知道如何建立協程的均可以調用 loadSortedProducts，例如 WorkManager 庫啓動了後臺任務。

Repository 應該使用掛起函數來保證主線程安全。

注意：當用戶離開界面時，一些後臺執行的保存操做可能想繼續運行，這種狀況下，脫離生命週期運行是有意義的。在大多數狀況下，viewModelScope 都是一個好選擇。

再來看看 ProductsDao ：

@Dao
interface ProductsDao {
    // Because this is marked suspend, Room will use it's own dispatcher
    // to run this query in a main-safe way,
    @Query("select * from ProductListing ORDER BY dataStocked ASC")
    suspend fun loadProductsByDateStockedAsceding(): List<ProductListing>
    
    // Because this is marked suspend, Room will use it's own dispatcher
    // to run this query in a main-safe way,
    @Query("select * from ProductListing ORDER BY dataStocked DESC")
    suspend fun loadProductsByDateStockedDesceding(): List<ProductListing>
}
複製代碼

ProductsDao 是一個 Room Dao，它對外提供了兩個掛起函數。因爲函數由 suspend 修飾，Room 會確保它們主線程安全。這就意味着你能夠直接在 Dispatchers.Main 中調用它們。

若是你沒有在 Room 中使用過協程，閱讀一下 FMuntenescu 的這篇文章:

Room && Coroutines

不過須要注意這一點，調用它的協程將運行在主線程。因此若是你要對結果進行一些昂貴的操做，例如轉換成集合，你要確保不會阻塞主線程。

注意：Room 使用本身的調度器在後臺線程進行查詢操做。你不該該再使用 withContext(Dispatchers.IO) 來調用 Room 的 suspend 查詢，這隻會讓你的代碼運行的更慢。

Room 中的掛起函數是主線程安全的，它運行在自定義的調度器中。

一次性請求模式

這就是在 Android Architecture Components 中使用協程進行一次性請求的完整模式。咱們將協程添加到 ViewModel 、Repository 和 Room 中，每一層都有不一樣的責任。

ViewModel 在主線程啓動協程，一旦有告終果就結束。
Repository 提供掛起函數並保證它們主線程安全。
數據庫和網絡層提供掛起函數並保證它們主線程安全。

ViewModel 負責啓動協程，保證用戶離開界面時取消協程。它自己不作昂貴的操做，而是依賴其餘層來作。一旦有告終果，就使用 LiveData 發送給 UI 界面。也正由於 ViewModel 不作昂貴的操做，因此它在主線程啓動協程。經過在主線程啓動，當結果可用它能夠更快的響應用戶事件（例如內存緩存）。

Repository 提供掛起函數來訪問數據。它一般不會啓動長生命週期的協程，由於它沒有辦法取消它們。不管什麼時候 Repository 須要作昂貴的操做（集合轉換等），它都須要使用 withContext 來提供主線程安全的接口。

數據層（網絡或者數據庫）老是提供掛起函數。使用 Kotlin 協程的時候須要保證這些掛起函數是主線程安全的，Room 和 Retrofit 都遵循了這一原則。

在一次性請求中，數據層只提供掛起函數。若是想要獲取新值，就必須再調用一次。這就像瀏覽器中的刷新按鈕。

花點時間讓你明白一次性請求的模式是值得的。這在 Android 協程中是通用的模式，你也會一直使用它。

第一個 Bug Report

在測試過該解決方案以後，你將其用到生產環境，幾周內都運行良好，直到你收到了一個很是奇怪的錯誤報告：

Subject: 🐞 — 排序錯誤！

Report: 當我很是很是很是很是快速點擊排序按鈕時，排序偶爾是錯誤的。這並非每次都會發生。

你看了看，撓撓頭，哪裏可能發生錯誤了呢？這個邏輯看起來至關簡單：

開始用戶請求的排序
在 Room 調度器中開始排序
展現排序結果

你正準備關閉這個 bug，關閉理由是「不予處理 —— 不要快速點擊按鈕」，可是你又擔憂的確是哪裏出了什麼問題。在添加了日誌以及編寫測試用例來測試一次性發起許多排序請求，你最終找到了緣由。

最後得到的結果實際上並非「排序的結果」，而是「完成最後一次排序時」的結果。當用戶狂點按鈕時，同時發起了屢次排序，可能以任意順序結束。（譯者注：能夠想象成 Java 中的多線程併發）

當啓動一個新協程來響應用戶事件時，要考慮到用戶在該協程未結束以前又啓動一個協程會發生什麼。

這是一個併發致使的 bug，實際上它和協程並無什麼關係。當咱們以一樣的方式使用回調，Rx，甚至 ExecutorService，均可能會有這樣的 bug。讓咱們探索一下下面這些方案是如何保證一次性請求按用戶所指望的順序執行的。

最佳方案：禁用按鈕

核心問題就是咱們如何進行兩次排序。咱們可讓它僅僅只進行一次排序！最簡單的方法就是禁用排序按鈕，中止發送新事件。

這彷佛是一個很簡單的方案，但它的確是個好主意。代碼實現也很簡單，易於測試。

要禁用按鈕，能夠通知 UI sortPricesBy 中正在進行一次排序請求，以下所示：

// Solution 0: Disable the sort buttons when any sort is running

class ProductsViewModel(val productsRepository: ProductsRepository): ViewModel() {
   private val _sortedProducts = MutableLiveData<List<ProductListing>>()
   val sortedProducts: LiveData<List<ProductListing>> = _sortedProducts
  
   private val _sortButtonsEnabled = MutableLiveData<Boolean>()
   val sortButtonsEnabled: LiveData<Boolean> = _sortButtonsEnabled
  
   init {
       _sortButtonsEnabled.value = true
   }

   /** * Called by the UI when the user clicks the appropriate sort button */
   fun onSortAscending() = sortPricesBy(ascending = true)
   fun onSortDescending() = sortPricesBy(ascending = false)

   private fun sortPricesBy(ascending: Boolean) {
       viewModelScope.launch {
           // disable the sort buttons whenever a sort is running
           _sortButtonsEnabled.value = false
           try {
               _sortedProducts.value =
                       productsRepository.loadSortedProducts(ascending)
           } finally {
               // re-enable the sort buttons after the sort is complete
               _sortButtonsEnabled.value = true
           }
       }
   }
}
複製代碼

這看起來還不賴。只需在調用 repository 時在 sortPricesBy 內部禁用按鈕。

大多數狀況下，這都是解決問題的好方案。可是咱們想在按鈕可用的狀況下來解決這個 bug 呢？這有一點點困難，咱們將在本文剩餘部分來看幾種方式。

Important ：This code shows a major advantage of starting on main — the buttons disable instantly in response to a click. If you switched dispatchers, a fast-fingered user on a slow phone could send more than one click!

併發模式

下面幾節將探討一些高級話題。若是你纔剛剛開始使用協程，你沒必要徹底理解。簡單的禁用按鈕就是你遇到的大部分問題的良好解決方案。

在本文的剩餘部分，咱們將討論在不由用按鈕的前提下，如何去保證一次性請求正常運行。咱們能夠經過控制協程什麼時候運行（或者不運行）來避免意外的併發狀況。

下面有三種模式，你能夠在一次性請求中使用它們來確保同一時間只進行一次請求。

在啓動更多協程以前先取消上一個。
將下一個任務放入等待隊列，直到前一個請求執行完成在開始另外一個。
若是已經有一個請求在運行，那麼就返回該請求，而不是啓動另外一個請求。

想一下這些解決方案，你會發現它們的實現相對都比較複雜。爲了專一於設計模式而不是實現細節，我建立了 gist 來提供這三種模式的實現做爲可用抽象。（這裏能夠大概瀏覽一下 gist 中的代碼實現）

方案一 : 取消前一個任務

在排序的狀況下，從用戶那獲取了一個新的事件，就意味着你能夠取消上一個請求了。畢竟，用戶已經不想知道上一個任務的結果了，繼續下去還有什麼意義呢？

爲了取消上一個請求，咱們首先要以某種方式追蹤它。gist 中的 cancelPreviousThenRun 函數就是這麼作的。

讓咱們看看它是如何被用來修復 bug 的：

// Solution #1: Cancel previous work

// This is a great solution for tasks like sorting and filtering that
// can be cancelled if a new request comes in.

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   var controlledRunner = ControlledRunner<List<ProductListing>>()

   suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
       // cancel the previous sorts before starting a new one
       return controlledRunner.cancelPreviousThenRun {
           if (ascending) {
               productsDao.loadProductsByDateStockedAscending()
           } else {
               productsDao.loadProductsByDateStockedDescending()
           }
       }
   }
}
複製代碼

看一下 gist 中 cancelPreviousThenRun 中的實現，你能夠了解到它是如何追蹤正在工做的任務的。

// see the complete implementation at
// https://gist.github.com/objcode/7ab4e7b1df8acd88696cb0ccecad16f7
suspend fun cancelPreviousThenRun(block: suspend () -> T): T {
   // If there is an activeTask, cancel it because it's result is no longer needed
   activeTask?.cancelAndJoin()
   
   // ...
複製代碼

簡而言之，它老是追蹤成員變量 activeTask 中的當前排序。不管什麼時候開始一次新的排序，都會當即 cancelAndJoin activeTask 中的全部內容。這會形成的影響就是，在開啓一次新的排序以前會取消全部正在進行的排序。

使用相似 ControlledRunner<T> 的抽象實現來封裝邏輯是個好方法，而不是將併發性和程序邏輯混雜在一塊兒。

重要：這個模式不適合在全局單例中使用，由於不相關的調用者不該該互相取消。

方案二：將下一個任務入隊

這裏有一個對於併發 bug 老是有效的解決方案。

只須要將請求排隊，這樣同時只會進行一個請求。就像商店中排隊同樣，請求將按它們排隊的順序依次執行。

對於這種特定的排隊問題，取消可能比排隊更好。但值得一提的是它老是能夠保證正常工做。

// Solution #2: Add a Mutex

// Note: This is not optimal for the specific use case of sorting
// or filtering but is a good pattern for network saves.

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   val singleRunner = SingleRunner()

   suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
       // wait for the previous sort to complete before starting a new one
       return singleRunner.afterPrevious {
           if (ascending) {
               productsDao.loadProductsByDateStockedAscending()
           } else {
               productsDao.loadProductsByDateStockedDescending()
           }
       }
   }
}
複製代碼

不管什麼時候進行一次新的排序，它使用一個 SingleRunner 實例來確保同時只進行一個排序任務。

它使用了 Mutex ，Mutex（互斥鎖）是一個單程票，或者說鎖，協程必須獲取鎖才能進入代碼塊。若是一個協程在運行時另外一個協程嘗試進入，它將掛起本身直到全部等待的協程都完成。

Mutex 保證同時只有一個協程運行，而且它們將按啓動的順序結束。

方案三：加入前一個任務

第三種解決方案是加入前一個任務。若是新請求能夠重複使用已經存在的，已經完成了一半的相同的任務，這會是一個好主意。

這種模式對於排序功能來講並無太大意義，可是對於網絡請求來講是很適用的。

對於咱們的產品庫存應用，用戶須要一種方式來從服務器獲取最新的產品庫存數據。咱們提供了一個刷新按鈕，用戶能夠點擊來發起一次新的網絡請求。

就和排序按鈕同樣，當請求正在進行的時候，禁用按鈕就能夠解決問題。可是若是咱們不想這樣，或者不能這樣，咱們能夠選擇加入已經存在的請求。

查看 gist 中使用 joinPreviousOrRun 的代碼，看看它是如何工做的：

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   var controlledRunner = ControlledRunner<List<ProductListing>>()

   suspend fun fetchProductsFromBackend(): List<ProductListing> {
       // if there's already a request running, return the result from the 
       // existing request. If not, start a new request by running the block.
       return controlledRunner.joinPreviousOrRun {
           val result = productsApi.getProducts()
           productsDao.insertAll(result)
           result
       }
   }
}
複製代碼

這與 cancelPreviousAndRun 的行爲相反。cancelPreviousAndRun 會經過取消直接放棄前一個請求，而 joinPreviousOrRun 將會放棄新請求。若是已經存在正在運行的請求，它將會等待執行結果並返回，而不是發起一次新的請求。只有在沒有正在運行的請求時纔會執行代碼塊。

在下面的代碼中你能夠看到 joinPreviousOrRun 中的任務是如何工做的。它僅僅只是當 activeTask 中存在任務的時候，直接返回前一個請求的結果。

// see the complete implementation at
// https://gist.github.com/objcode/7ab4e7b1df8acd88696cb0ccecad16f7#file-concurrencyhelpers-kt-L124

suspend fun joinPreviousOrRun(block: suspend () -> T): T {
    // if there is an activeTask, return it's result and don't run the block
    activeTask?.let {
        return it.await()
    }
    // ...
複製代碼

這個模式很適合經過 id 查詢產品的請求。你可使用 map 來保存 id 到 Deferred 的映射關係，而後使用相同的 join 邏輯來追蹤同一個產品的以前的請求。