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

單向數據流

△加載數據流的過程

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

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

△山上的 Pancho

但有時 Pahcho 不走運，走到湖邊時發現湖水已經干涸，于是就不得不再去別處尋找水源。發生了幾次這種情況后，Pancho 意識到，搭建一些基礎設施可以解決這個問題。于是他在湖邊安裝了一些管道，當湖中有水時，只用擰開水龍頭就能取到水。知道了如何安裝管道，就能很自然地想到從多個水源地把管道組合，這樣一來 Pancho 就不必再檢查湖水是否已經干涸。

△鋪設管道

在 Android 應用中您可以簡單地在每次需要時請求數據，例如我們可以使用掛起函數來實現在每次視圖啟動時向 ViewModel 請求數據，而后 ViewModel 又向數據層請求數據，接下來這一切又在相反的方向上發生。不過這樣過了一段時間之后，像 Pancho 這樣的開發者們往往會想到，其實有必要投入一些成本來構建一些基礎設施，我們就可以不再請求數據而改為觀察數據。觀察數據就像安裝取水管道一樣，部署完成后對數據源的任何更新都將自動向下流動到視圖中，Pancho 再也不用走到湖邊去了。
△傳統的請求數據與單向數據流

響應式編程

我們將這類觀察者會自動對被觀察者對象的變化而作出反應的系統稱之為響應式編程，它的另一個設計要點是保持數據只在一個方向上流動，因為這樣更容易管理且不易出錯。 某個示例應用界面的 "數據流動" 如下圖所示，身份認證管理器會告訴數據庫用戶已登錄，而數據庫又必須告訴遠程數據源來加載一組不同的數據；與此同時這些操作在獲取新數據時都會告訴視圖顯示一個轉圈的加載圖標。對此我想說這雖然是可行的，但容易出現錯誤。

△錯綜復雜的 "數據流動"

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

△單向數據綁定

使用 Flow

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

△生產者和消費者

生產者會將數據 emit (發送) 到數據流中，而消費者則從數據流中 collect (收集) 這些數據。在 Android 中數據源或存儲區通常是應用數據的生產者；消費者則是視圖，它會把數據顯示在屏幕上。

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

△提供 Flow 支持的庫

我們來看一個 Room 的例子。您可以通過導出指定類型的數據流來獲取數據庫中發生變更的通知。在本例中，Room 庫是生產者，它會在每次查詢后發現有更新時發送內容。

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

創建 Flow

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

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

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

轉換 Flow

在 Android 中，生產者和消費者之間的層可以使用中間運算符修改數據流來適應下一層的要求。

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

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())}

收集 Flow

現在我們已經了解過如何生成和修改數據流，接下來了解一下如何收集數據流。收集數據流通常發生在視圖層，因為這是我們想要在屏幕上顯示數據的地方。

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

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

在 Flow 中使用終端運算符將按需創建數據流并開始發送值，而相反的是中間操作符只是設置了一個操作鏈，其會在數據被發送到數據流時延遲執行。每次對 userMessages 調用 collect 時都會創建一個新的數據流，其生產者代碼塊將根據自己的時間間隔開始刷新來自 API 的消息。在協程中我們將這種按需創建并且只有在被觀察時才會發送數據的數據流稱之為冷流 (Cold Stream)。

在 Android 視圖上收集數據流

在 Android 的視圖中收集數據流要注意兩點，第一是在后臺運行時不應浪費資源，第二是配置變更。

安全收集

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

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

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)                    }                }                // 協程將會在 lifecycle 進入 DESTROYED 后被恢復            }    }}

repeatOnLifecycle 的可重啟行為充分考慮了界面的生命周期，不過需要注意的是，直到生命周期進入 DESTROYED，調用 repeatOnLifecycle 的協程都不會恢復執行，因此如果您需要從多個數據流中進行收集，則應在 repeatOnLifecycle 代碼塊內多次使用 launch 來創建協程:

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

如果只需從一個數據流中進行收集，則可使用 flowWithLifecycle 來收集數據，它能夠在生命周期進入目標狀態時發送數據，并在離開目標狀態時取消內部的生產者:

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

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

△Activity 的生命周期

repeatOnLifecycle 和 flowWithLifecycle 是 lifecycle-runtime-ktx 庫在 2.4.0 穩定版中新增的 API，在沒有這些 API 之前您可能已經以其他方式從 Android 界面中收集數據流，例如像上面的代碼一樣直接從 lifecycleScope.launch 啟動的協程中收集，雖然這樣看起來也能工作但不一定安全，因為這種方式將持續從數據流中收集數據并更新界面元素，即便是應用退出到后臺時也一樣。如果使用 launchWhenStarted 替代它的話，情況會稍微好一些，因為它會在處于后臺時將收集掛起。但這樣會在讓數據流生產者保持活躍狀態，有可能會在后臺持續發出不需要在屏幕上顯示的數據項，從而將內存占滿。由于界面并不知道數據流生產者的實現方式，所以最好謹慎一些，使用 repeatOnLifecycle 或 flowWithLifecycle 來避免界面在處于后臺時收集數據或保持數據流生產者處于活躍狀態。 下面是一段不安全的使用方式示例:

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 { … }            }    }}

配置變更

當您向視圖暴露數據流時，必須要考慮到您正在嘗試在具有不同生命周期的兩個元素之間傳遞數據，并不是所有生命周期都會出現問題，但在 Activity 和 Fragment 的生命周期里會比較棘手。當設備旋轉或者接收到配置變更時，所有的 Activity 都可能會重啟但 ViewModel 卻能被保留，因此您不能把任意數據流都簡單地從 ViewModel 中暴露出來。

△旋轉屏幕會重建 Activity 但能夠保留 ViewModel

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

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

您可以使用 StateFlow 的可變版本，并隨時根據需要在協程中更新它的值，但這樣做可能不太符合響應式編程的風格，如下代碼所示:

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

Pancho 會建議您將各種類型的數據流都轉換為 StateFlow 來改進這個問題，這樣 StateFlow 將接收來自上游數據流的所有更新并存儲最新的值，并且收集器的數量可以是 0 至任意多個，因此非常適合與 ViewModel 一起使用。當然，除此之外還有一些其他類型的 Flow，但推薦您使用 StateFlow，因為我們可以對它進行非常精確的優化。

△將任意數據流轉換為 StateFlow

要將數據流轉換為 StateFlow 可以使用 stateIn 運算符，它需要傳入三個參數: initinalValue、scope 及 started。其中 initialValue 是因為 StateFlow 必須有值；而協程 scope 則是用于控制何時開始共享，在上面的例子中我們使用了 viewModelScope；最后的 started 是個有趣的參數，我們后面會聊到 WhileSubscribed(5000) 的作用，先看這部分的代碼:

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

我們來看看這兩個場景: 第一種場景是旋轉，在該場景中 Activity (也就是數據流收集器) 在短時間內被銷毀然后重建；第二個場景是回到主屏幕，這將會使我們的應用進入后臺。在旋轉場景中我們不希望重啟任何數據流以便盡可能快地完成過渡，而在回到主屏幕的場景中我們則希望停止所有數據流以便節省電量和其他資源。

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

△設置超時時間來應對不同的場景

總的來說，建議您使用 StateFlow 來通過 ViewModel 暴露數據流，或者使用 asLiveData 來實現同樣的目的，關于 StateFlow 或其父類 SharedFlow 的更多詳細信息，請參閱:StateFlow 和 SharedFlow。

測試數據流

測試數據流可能會比較復雜，因為要處理的對象是流式數據，這里介紹在兩個不同的場景中有用的小技巧: 首先是第一個場景，被測單元依賴了數據流，那對此類場景進行測試最簡單的方法就是用模擬生產者替代依賴項。在本例中，您可以對這個模擬源進行編程以對不同的測試用例發送其所需要的內容。您可以像上面的例子一樣實現一個簡單的冷流，測試本身會對受測對象的輸出進行斷言，輸出的內容可以是數據流或其他任何類型。

△被測單元依賴數據流的測試技巧

模擬被測單元所依賴的數據流:

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

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

△測試數據流的技巧

測試數據流:

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

回顧

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

原文標題：實戰 | 使用 Kotlin Flow 構建數據流 "管道"