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

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

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

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

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

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

響應(yīng)式編程

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

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

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

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

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

大多數(shù)情況下您都無需自行創(chuàng)建數(shù)據(jù)流，因?yàn)閿?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 中，當(dāng)您希望頻繁地在應(yīng)用內(nèi)檢查新消息時(shí)，可以將用戶消息暴露為消息列表類型的數(shù)據(jù)流。我們使用數(shù)據(jù)流構(gòu)建器來創(chuàng)建數(shù)據(jù)流，因?yàn)?Flow 是在協(xié)程上下文環(huán)境中運(yùn)行的，它以掛起代碼塊作為參數(shù)，這也意味著它能夠調(diào)用掛起函數(shù)，我們可以在代碼塊中使用 while(true) 來循環(huán)執(zhí)行我們的邏輯。

在示例代碼中，我們首先從 API 獲取消息，然后使用 emit 掛起函數(shù)將結(jié)果添加到 Flow 中，這將掛起協(xié)程直到收集器接收到數(shù)據(jù)項(xiàng)，最后我們將協(xié)程掛起一段時(shí)間。在 Flow 中，操作會在同一個(gè)協(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) //  掛起一段時(shí)間        }    }}

轉(zhuǎn)換 Flow

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

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

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ā)生在視圖層，因?yàn)檫@是我們想要在屏幕上顯示數(shù)據(jù)的地方。

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

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

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

安全收集

假設(shè)我們在 MessagesActivity 中，如果希望在屏幕上顯示消息列表，則應(yīng)該當(dāng)界面沒有顯示在屏幕上時(shí)停止收集，就像是 Pancho 在刷牙或者睡覺時(shí)應(yīng)該關(guān)上水龍頭一樣。我們有多種具有生命周期感知能力的方案，來實(shí)現(xiàn)當(dāng)信息不在屏幕上展示就不從數(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 來實(shí)現(xiàn)這件事情。不過為了簡單起見，這里推薦使用 repeatOnLifecycle 從界面層收集數(shù)據(jù)流。

repeatOnLifecycle 是一個(gè)接收 Lifecycle.State 作為參數(shù)的掛起函數(shù)，該 API 具有生命周期感知能力，所以能夠在當(dāng)生命周期進(jìn)入響應(yīng)狀態(tài)時(shí)自動使用傳遞給它的代碼塊啟動新的協(xié)程，并且在生命周期離開該狀態(tài)時(shí)取消該協(xié)程。在上面的例子中，我們使用了 Activity 的 lifecycleScope 來啟動協(xié)程，由于 repeatOnLifecycle 是掛起函數(shù)，所以它需要在協(xié)程中被調(diào)用。最佳實(shí)踐是在生命周期初始化時(shí)調(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 進(jìn)入 DESTROYED 后被恢復(fù)            }    }}

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

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

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

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

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

△Activity 的生命周期

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

配置變更

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

以如下代碼中的冷流為例，由于每次收集冷流時(shí)它都會重啟，所以在設(shè)備旋轉(zhuǎn)之后會再次調(diào)用 repository.fetchItem()。我們需要某種緩沖區(qū)機(jī)制來保障無論重新收集多少次都可以保持?jǐn)?shù)據(jù)，并在多個(gè)收集器之間共享數(shù)據(jù)，而 StateFlow 正是為了此用途而設(shè)計(jì)的。在我們的湖泊比喻中，StateFlow 就好比水箱，即使沒有收集器它也能持有數(shù)據(jù)。因?yàn)樗梢远啻伪皇占阅軌蚍判牡貙⑵渑c 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è)置超時(shí)時(shí)間來正確判斷不同的場景，當(dāng)停止收集 StateFlow 時(shí)，不會立即停止所有上游數(shù)據(jù)流，而是會等待一段時(shí)間，如果在超時(shí)前再次收集數(shù)據(jù)則不會取消上游數(shù)據(jù)流，這就是 WhileSubscribed(5000) 的作用。當(dāng)設(shè)置了超時(shí)時(shí)間后，如果按下主屏幕按鈕會讓視圖立即結(jié)束收集，但 StateFlow 會經(jīng)過我們設(shè)置的超時(shí)時(shí)間之后才會停止其上游數(shù)據(jù)流，如果用戶再次打開應(yīng)用則會自動重啟上游數(shù)據(jù)流。而在旋轉(zhuǎn)場景中視圖只停止了很短的時(shí)間，無論如何都不會超過 5 秒鐘，因此 StateFlow 并不會重啟，所有的上游數(shù)據(jù)流都將會保持在活躍狀態(tài)，就像什么都沒有發(fā)生一樣可以做到即時(shí)向用戶呈現(xiàn)旋轉(zhuǎn)后的屏幕。

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

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

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

@Testfun myTest() = runBlocking {     // 收集第一個(gè)數(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)容的詳細(xì)信息，我們在早些時(shí)候?yàn)槠涓铝撕芏嘤嘘P(guān)數(shù)據(jù)流的內(nèi)容。

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