響應(yīng)式編程在Android開發(fā)中的實踐

Flow 是一種基于流的編程模型，本文我們將向大家介紹響應(yīng)式編程以及其在 Android 開發(fā)中的實踐，您將了解到如何將生命周期、旋轉(zhuǎn)及切換到后臺等狀態(tài)綁定到 Flow 中，并且測試它們是否能按照預(yù)期執(zhí)行。

單向數(shù)據(jù)流

△加載數(shù)據(jù)流的過程

每款 Android 應(yīng)用都需要以某種方式收發(fā)數(shù)據(jù)，比如從數(shù)據(jù)庫獲取用戶名、從服務(wù)器加載文檔，以及對用戶進行身份驗證等。接下來，我們將介紹如何將數(shù)據(jù)加載到 Flow，然后經(jīng)過轉(zhuǎn)換后暴露給視圖進行展示。

為了大家更方便地理解 Flow，我們以 Pancho (潘喬) 的故事來展開。當住在山上的 Pancho 想從湖中獲取淡水時，會像大多數(shù)新手一開始一樣，拿個水桶走到湖邊取水，然后再走回來。

響應(yīng)式編程

更好的方式則是讓數(shù)據(jù)只在一個方向上流動，并創(chuàng)建一些基礎(chǔ)設(shè)施 (像 Pancho 鋪設(shè)管道那樣) 來組合和轉(zhuǎn)換這些數(shù)據(jù)流，這些管道可以隨著狀態(tài)的變化而修改，比如在用戶退出登錄時重新安裝管道。

△單向數(shù)據(jù)綁定

可以想象對于這些組合和轉(zhuǎn)換來說，我們需要一個成熟的工具來完成這些操作。在本文中我們將使用 Kotlin Flow 來實現(xiàn)。Flow 并不是唯一的數(shù)據(jù)流構(gòu)建器，不過得益于它是協(xié)程的一部分并且得到了很好的支持。我們剛才一直用作比喻的水流，在協(xié)程庫里稱之為 Flow 類型，我們用泛形 T 來指代數(shù)據(jù)流承載的用戶數(shù)據(jù)或者頁面狀態(tài)等任何類型。

△生產(chǎn)者和消費者

生產(chǎn)者會將數(shù)據(jù) emit (發(fā)送) 到數(shù)據(jù)流中，而消費者則從數(shù)據(jù)流中 collect (收集) 這些數(shù)據(jù)。在 Android 中數(shù)據(jù)源或存儲區(qū)通常是應(yīng)用數(shù)據(jù)的生產(chǎn)者；消費者則是視圖，它會把數(shù)據(jù)顯示在屏幕上。

大多數(shù)情況下您都無需自行創(chuàng)建數(shù)據(jù)流，因為數(shù)據(jù)源中依賴的庫，例如 DataStore、Retrofit、Room 或 WorkManager 等常見的庫都已經(jīng)與協(xié)程及 Flow 集成在一起了。這些庫就像是水壩，它們使用 Flow 來提供數(shù)據(jù)，您無需了解數(shù)據(jù)是如何生成的，只需 "接入管道" 即可。

△提供 Flow 支持的庫

@DAOinterface CodelabsDAO {     @Query("SELECT * FROM codelabs")    fun getAllCodelabs(): Flow>}

如果您要自己創(chuàng)建數(shù)據(jù)流，有一些方案可供選擇，比如數(shù)據(jù)流構(gòu)建器。假設(shè)我們處于 UserMessagesDataSource 中，當您希望頻繁地在應(yīng)用內(nèi)檢查新消息時，可以將用戶消息暴露為消息列表類型的數(shù)據(jù)流。我們使用數(shù)據(jù)流構(gòu)建器來創(chuàng)建數(shù)據(jù)流，因為 Flow 是在協(xié)程上下文環(huán)境中運行的，它以掛起代碼塊作為參數(shù)，這也意味著它能夠調(diào)用掛起函數(shù)，我們可以在代碼塊中使用 while(true) 來循環(huán)執(zhí)行我們的邏輯。

在示例代碼中，我們首先從 API 獲取消息，然后使用 emit 掛起函數(shù)將結(jié)果添加到 Flow 中，這將掛起協(xié)程直到收集器接收到數(shù)據(jù)項，最后我們將協(xié)程掛起一段時間。在 Flow 中，操作會在同一個協(xié)程中順序執(zhí)行，使用 while(true) 循環(huán)可以讓 Flow 持續(xù)獲取新消息直到觀察者停止收集數(shù)據(jù)。傳遞給數(shù)據(jù)流構(gòu)建器的掛起代碼塊通常被稱為 "生產(chǎn)者代碼塊"。

class UserMessagesDataSource(    private val messagesApi: MessagesApi,    private val refreshIntervalMs: Long = 5000) {    val latestMessages: Floa> = flow {        white(true) {            val userMessages = messagesApi.fetchLatestMessages()            emit(userMessages) // 將結(jié)果發(fā)送給 Flow            delay(refreshIntervalMs) //  掛起一段時間        }    }}

轉(zhuǎn)換 Flow

在 Android 中，生產(chǎn)者和消費者之間的層可以使用中間運算符修改數(shù)據(jù)流來適應(yīng)下一層的要求。

在本例中，我們將 latestMessages 流作為數(shù)據(jù)流的起點，則可以使用 map 運算符將數(shù)據(jù)轉(zhuǎn)換為不同的類型，例如我們可以使用 map lambda 表達式將來自數(shù)據(jù)源的原始消息轉(zhuǎn)換為 MessagesUiModel，這一操作可以更好地抽象當前層級，每個運算符都應(yīng)根據(jù)其功能創(chuàng)建一個新的 Flow 來發(fā)送數(shù)據(jù)。我們還可以使用 filter 運算符過濾數(shù)據(jù)流來獲得包含重要通知的數(shù)據(jù)流。而 catch 運算符則可以捕獲上游數(shù)據(jù)流中發(fā)生的異常，上游數(shù)據(jù)流是指在生產(chǎn)者代碼塊和當前運算符之間調(diào)用的運算符產(chǎn)生的數(shù)據(jù)流，而在當前運算符之后生成的數(shù)據(jù)流則被稱為下游數(shù)據(jù)流。catch 運算符還可以在有需要的時候再次拋出異常或者發(fā)送新值，我們在示例代碼中可以看到其在捕獲到 IllegalArgumentExceptions 時將其重新拋出，并且在發(fā)生其他異常時發(fā)送一個空列表:

val importantUserMessages: Flow<MessageUiModel> =     userMessageDataSource.latestMessages        .map { userMessage ->            userMessages.toUiModel()        }        .filter { messageUiModel ->            messagesUiModel.containsImportantNotifications()        }        .catch { e ->            analytics.log("Error loading reserved event")            if (e is IllegalArgumentException) throw e            else emit(emptyList())}

現(xiàn)在我們已經(jīng)了解過如何生成和修改數(shù)據(jù)流，接下來了解一下如何收集數(shù)據(jù)流。收集數(shù)據(jù)流通常發(fā)生在視圖層，因為這是我們想要在屏幕上顯示數(shù)據(jù)的地方。

在本例中，我們希望列表中能夠顯示最新消息以便 Pancho 能夠了解最新動態(tài)。我們可以使用終端運算符 collect 來監(jiān)聽數(shù)據(jù)流發(fā)送的所有值，collect 接收一個函數(shù)作為參數(shù)，每個新值都會調(diào)用該參數(shù)，并且由于它是一個掛起函數(shù)，因此需要在協(xié)程中執(zhí)行。

userMessages.collect{messages->    listAdapter.submitList(messages)}

在 Android 視圖上收集數(shù)據(jù)流

安全收集

假設(shè)我們在 MessagesActivity 中，如果希望在屏幕上顯示消息列表，則應(yīng)該當界面沒有顯示在屏幕上時停止收集，就像是 Pancho 在刷牙或者睡覺時應(yīng)該關(guān)上水龍頭一樣。我們有多種具有生命周期感知能力的方案，來實現(xiàn)當信息不在屏幕上展示就不從數(shù)據(jù)流中收集信息的功能，比如 androidx.lifecycle:lifecycle-runtime-ktx 包中的 Lifecycle.repeatOnLifecycle(state) 和 Flow.flowWithLifecycle(lifecycle, state)。您還可以在 ViewModel 中使用 androidx.lifecycle:lifecycle-livedata-ktx 包里的 Flow.asLiveData(): LiveData 將數(shù)據(jù)流轉(zhuǎn)換為 LiveData，這樣就可以像往常一樣使用 LiveData 來實現(xiàn)這件事情。不過為了簡單起見，這里推薦使用 repeatOnLifecycle 從界面層收集數(shù)據(jù)流。

repeatOnLifecycle 是一個接收 Lifecycle.State 作為參數(shù)的掛起函數(shù)，該 API 具有生命周期感知能力，所以能夠在當生命周期進入響應(yīng)狀態(tài)時自動使用傳遞給它的代碼塊啟動新的協(xié)程，并且在生命周期離開該狀態(tài)時取消該協(xié)程。在上面的例子中，我們使用了 Activity 的 lifecycleScope 來啟動協(xié)程，由于 repeatOnLifecycle 是掛起函數(shù)，所以它需要在協(xié)程中被調(diào)用。最佳實踐是在生命周期初始化時調(diào)用該函數(shù)，就像上面的例子中我們在 Activity 的 onCreate 中調(diào)用一樣:

import androidx.lifecycle.repeatOnLifecycle class MessagesActivity : AppCompatActivity() {     val viewModel: MessagesViewModel by viewModels()     override fun onCreate(savedInstanceState: Bundle?) {                       lifecycleScope.launch {                repeatOnLifecycle(Lifecycle.State.STARTED)                    viewModel.userMessages.collect { messages ->                        listAdapter.submitList(messages)                    }                }                // 協(xié)程將會在 lifecycle 進入 DESTROYED 后被恢復(fù)            }    }}

repeatOnLifecycle 的可重啟行為充分考慮了界面的生命周期，不過需要注意的是，直到生命周期進入 DESTROYED，調(diào)用 repeatOnLifecycle 的協(xié)程都不會恢復(fù)執(zhí)行，因此如果您需要從多個數(shù)據(jù)流中進行收集，則應(yīng)在 repeatOnLifecycle 代碼塊內(nèi)多次使用 launch 來創(chuàng)建協(xié)程:

lifecycleScope.launch {    repeatOnLifecycle(Lifecycle.State.STARTED) {             launch {                viewModel.userMessages.collect { … }            }             launch {                otherFlow.collect { … }            }    }}

如果只需從一個數(shù)據(jù)流中進行收集，則可使用 flowWithLifecycle 來收集數(shù)據(jù)，它能夠在生命周期進入目標狀態(tài)時發(fā)送數(shù)據(jù)，并在離開目標狀態(tài)時取消內(nèi)部的生產(chǎn)者:

lifecycleScope.launch {    viewModel.userMessages        .flowWithLifecycle(lifecycle, State.STARTED)        .collect { messages ->            listAdapter.submitList(messages)        }}

為了能夠直觀地展示具體的運作過程，我們來探索一下此 Activity 的生命周期，首先是創(chuàng)建完成并向用戶可見；接下來用戶按下了主屏幕按鈕將應(yīng)用退到后臺，此時 Activity 會收到 onStop 信號；當重新打開應(yīng)用時又會調(diào)用 onStart。如果您調(diào)用 repeatOnLifecycle 并傳入 STARTED 狀態(tài)，界面就只會在屏幕上顯示時收集數(shù)據(jù)流發(fā)出的信號，并且在應(yīng)用轉(zhuǎn)到后臺時取消收集。

△Activity 的生命周期

class MessagesActivity : AppCompatActivity() {     val viewModel: MessagesViewModel by viewModels()     override fun onCreate(savedInstanceState: Bundle?) {             //  危險的操作            lifecycleScope.launch {                viewModel.userMessage.collect { messages ->                    listAdapter.submitList(messages)                }            }             //  危險的操作            LifecycleCoroutineScope.launchWhenX {                flow.collect { … }            }    }}

配置變更

△旋轉(zhuǎn)屏幕會重建 Activity 但能夠保留 ViewModel

以如下代碼中的冷流為例，由于每次收集冷流時它都會重啟，所以在設(shè)備旋轉(zhuǎn)之后會再次調(diào)用 repository.fetchItem()。我們需要某種緩沖區(qū)機制來保障無論重新收集多少次都可以保持數(shù)據(jù)，并在多個收集器之間共享數(shù)據(jù)，而 StateFlow 正是為了此用途而設(shè)計的。在我們的湖泊比喻中，StateFlow 就好比水箱，即使沒有收集器它也能持有數(shù)據(jù)。因為它可以多次被收集，所以能夠放心地將其與 Activity 或 Fragment 一起使用。

val result: Flow>> = flow {    emit(repository.fetchItem())}

private val _myUiState = MutableStateFlow() val myUiState: StateFlow = _myUiState init {    viewModelScope.launch {        _muUiState.value = Result.Loading        _myUiState.value = repository.fetchStuff()    }}

val result: StateFlow> = someFlow    .stateIn(        initialValue = Result.Loading        scope = viewModelScope,        started = WhileSubscribed(5000),)

我們可以通過設(shè)置超時時間來正確判斷不同的場景，當停止收集 StateFlow 時，不會立即停止所有上游數(shù)據(jù)流，而是會等待一段時間，如果在超時前再次收集數(shù)據(jù)則不會取消上游數(shù)據(jù)流，這就是 WhileSubscribed(5000) 的作用。當設(shè)置了超時時間后，如果按下主屏幕按鈕會讓視圖立即結(jié)束收集，但 StateFlow 會經(jīng)過我們設(shè)置的超時時間之后才會停止其上游數(shù)據(jù)流，如果用戶再次打開應(yīng)用則會自動重啟上游數(shù)據(jù)流。而在旋轉(zhuǎn)場景中視圖只停止了很短的時間，無論如何都不會超過 5 秒鐘，因此 StateFlow 并不會重啟，所有的上游數(shù)據(jù)流都將會保持在活躍狀態(tài)，就像什么都沒有發(fā)生一樣可以做到即時向用戶呈現(xiàn)旋轉(zhuǎn)后的屏幕。

測試數(shù)據(jù)流

class MyFakeRepository : MyRepository {    fun observeCount() = flow {        emit(ITEM_1)    }}

如果受測單元暴露一個數(shù)據(jù)流，并且您希望驗證該值或一系列值，那么您可以通過多種方式收集它們。您可以對數(shù)據(jù)流調(diào)用 first() 方法以進行收集并在接收到第一個數(shù)據(jù)項后停止收集。您還可以調(diào)用 take(5) 并使用 toList 終端操作符來收集恰好 5 條消息，這種方法可能非常有幫助。

@Testfun myTest() = runBlocking {     // 收集第一個數(shù)據(jù)然后停止收集    val firstItem = repository.counter.first()     // 收集恰好 5 條消息    val first = repository.messages.take(5).toList()}

感謝閱讀本文，希望您通過本文內(nèi)容已經(jīng)了解到為什么響應(yīng)式架構(gòu)值得投資，以及如何使用 Kotlin Flow 構(gòu)建您的基礎(chǔ)設(shè)施。文末提供了有關(guān)這方面的資料，包括涵蓋基礎(chǔ)知識的指南以及深入探討某些主題的文章。另外您還可以通過 Google I/O 應(yīng)用了解這些內(nèi)容的詳細信息，我們在早些時候為其更新了很多有關(guān)數(shù)據(jù)流的內(nèi)容。

原文標題：實戰(zhàn) | 使用 Kotlin Flow 構(gòu)建數(shù)據(jù)流 "管道"