协程高级之结合Kotlin Flow与LiveData一起使用

2024-02-25 08:54:16

协程高级之结合Kotlin Flow与LiveData一起使用

1.前言

本文中，你将学习如何将Kotlin协程结合LiveData一起使用。
我们也将使用协程异步Flow来做同样的事情，这是一个协程Lib代表一个异步序列，或者流。

LiveData介绍及使用
协程异步Flow

我们将一个简单的sample app为例来进行展开，该app使用Android架构组件
构造，使用LiveData从Room数据库获取对象列表，并展示在RecyclerView网格布局中。

查询Room数据库的代码片段如下：

val plants: LiveData<List<Plant>> = plantDao.getPlants()

使用LiveData构造器和协程进行排序逻辑，LiveData可以自动更新

val plants: LiveData<List<Plant>> = liveData<List<Plant>> {
   val plantsLiveData = plantDao.getPlants()
   val customSortOrder = plantsListSortOrderCache.getOrAwait()
   emitSource(plantsLiveData.map { plantList -> plantList.applySort(customSortOrder) })
}

你也可以使用Flow实现

private val customSortFlow = plantsListSortOrderCache::getOrAwait.asFlow()

val plantsFlow: Flow<List<Plant>>
   get() = plantDao.getPlantsFlow()
       .combine(customSortFlow) { plants, sortOrder ->
           plants.applySort(sortOrder)
       }
       .flowOn(defaultDispatcher)
       .conflate()

学习前准备

  • 有Android架构组件使用经验，例如ViewModel, LiveData, Repository, 和 Room
  • 具备Kotlin基本语法技能，包括扩展函数和lambda表达式
  • Kotlin协程使用经验
  • Android中线程基本理解，包括主线程，后台线程及回调

Android架构组建使用介绍
Kotlin语法介绍
Kotlin协程介绍

内容概述

  • 转换LiveData为对Kotlin协程友好的LiveData构造器
  • LiveData构造器中添加业务逻辑
  • 使用Flow进行异步操作
  • 组合Flows并转换多个异步数据源
  • Flows的并发控制
  • 学习如何在LiveData和Flow之间进行选择

AS版本要求
要求AS版本3.5以上

2. 准备

下载Sample代码

$ git clone https://github.com/googlecodelabs/kotlin-coroutines.git

本文将使用到advanced-coroutines-codelab目录下的代码advanced-coroutines-codelab包含了几个模块，各模块功能如下：

  • start改造前的代码
  • finished_code使用协程改造后的代码
  • sunflower 业务数据提供模块

3. 运行sample App

运行start 模块，效果如下：
协程高级之结合Kotlin Flow与LiveData一起使用
植物园网格图
协程高级之结合Kotlin Flow与LiveData一起使用
植物过滤网格图

每种Plant都有一个growZoneNumber，代表每种植物最想生长的区域。所以用户可以使用这个属性对Plant进行过滤。

架构介绍

这个sample app使用了 Android架构组件将UI（MainActivity与PlantListFragment中的内容）与业务逻辑（PlantListViewModel中的内容）进行分离。PlantRepository为ViewModel与PlantDao提供桥梁，它可以访问Room数据库并返回Plant对象列表。UI层获取plants列表并显示在RecyclerView的网格布局中。

小提示：
Repository是ViewModel与Data之间的桥梁除了作为桥梁，repository可以被任何ViewModel访问。它也可以将多个数据源做逻辑合并，我们将在后面的章节对其做实现。

在代码修改之前，让我们快速浏览下，数据流如何从数据库流向UI。下面的代码片段展示了如何从ViewModel中加载plants列表:

PlantListViewModel.kt

val plants: LiveData<List<Plant>> = growZone.switchMap { growZone ->
    if (growZone == NoGrowZone) {
        plantRepository.plants
    } else {
        plantRepository.getPlantsWithGrowZone(growZone)
    }
}

GrowZone是一个内联类，仅有一个Int属性代表了区域Id。NoGrowZone代表了没有区域Id，仅用于过滤。

Plant.kt

inline class GrowZone(val number: Int)
val NoGrowZone = GrowZone(-1)

当点击过滤按钮时，growZone会进行切换。我们使用switchMap操作符决定返回的plants列表。

switchMap操作符号是LiveData操作符之一，用于数据变换。

下面的代码段展示了DAO操作，从数据库获取plant数据：

PlantDao.kt

@Query("SELECT * FROM plants ORDER BY name")
fun getPlants(): LiveData<List<Plant>>

@Query("SELECT * FROM plants WHERE growZoneNumber = :growZoneNumber ORDER BY name")
fun getPlantsWithGrowZoneNumber(growZoneNumber: Int): LiveData<List<Plant>>

PlantRepository.kt

val plants = plantDao.getPlants()
fun getPlantsWithGrowZone(growZone: GrowZone) =
    plantDao.getPlantsWithGrowZoneNumber(growZone.number)

由于大部分代码修改将集中在PlantListViewModel和PlantRepository，花一点时间来熟悉工程的结构是很有必要的，重点关注数据在软件层上的传递过程，
从database到Fragment。接下来，我们将使用LiveData构造器来改造plants列表的排序。

小提示:
本文基于Android Sunflower示例工程。

4. Plants的自定义排序

当前的Plants是按照字母排序的，我们想让特定的Plants排在前面，然后剩余的按字母排序。有点像购物app，会把提供了广告赞助的商品排在前面。我们的产品团队想要动态地更改排序而不需要重新发布版本，
因此我们将从后端获取plants列表并进行排序。

协程高级之结合Kotlin Flow与LiveData一起使用自定义排序效果图

自定义排序列表包含4种Plants：Orange, Sunflower, Grape, and Avocado.注意看它们是如何显示在列表前端的，然后剩余的plants按照字母排序。

现在按一下过滤按钮（仅有GrowZone Id为9 plants被显示出来）。Avocado从列表消失，由于它不属于GrowZone Id为9的Plants。另外4种Plants在GrowZone Id为9的列表中，因此，它们将保留在列表顶端。
协程高级之结合Kotlin Flow与LiveData一起使用
过滤后的排序列表效果图

下面让我们开始自定义排序的实现

5. 获取排列顺序

我们开始编写挂起函数用来从网络获取自定义排序列表，并缓存到内存中。

PlantRepository.kt

private var plantsListSortOrderCache = 
    CacheOnSuccess(onErrorFallback = { listOf<String>() }) {
        plantService.customPlantSortOrder()
    }

plantsListSortOrderCache用于自定义列表的内存缓存。如果有网络错误，它将返回一个空列表，因此，我们的app仍然可以展示数据，尽管没有获取到排序的列表。

这段代码使用CacheOnSuccess工具类处理缓存。通过抽象方式实现缓存细节，app可以直接使用它。由于CacheOnSuccess已经测试过了，我们不需要写很多的测试代码来验证API的可靠性。
当使用kotlinx-coroutines时，在你的代码中引入高级抽象是不错的主意。

现在让我们结合一些逻辑将排序应用于植物列表。

PlantRepository.kt

private fun List<Plant>.applySort(customSortOrder: List<String>): List<Plant> {
    return sortedBy { plant ->
        val positionForItem = customSortOrder.indexOf(plant.plantId).let { order ->
            if (order > -1) order else Int.MAX_VALUE
        }
        ComparablePair(positionForItem, plant.name)
    }
}

这个扩展函数将重排列表，将customSortOrder中的植物放到列表前端。

6. 使用LiveData构建逻辑

使用LiveData构造器改造plants与getPlantsWithGrowZone的代码如下：
PlantRepository.kt

val plants: LiveData<List<Plant>> = liveData<List<Plant>> {
   val plantsLiveData = plantDao.getPlants()
   val customSortOrder = plantsListSortOrderCache.getOrAwait()
   emitSource(plantsLiveData.map {
       plantList -> plantList.applySort(customSortOrder) 
   })
}

fun getPlantsWithGrowZone(growZone: GrowZone) = liveData {
    val plantsGrowZoneLiveData = plantDao.getPlantsWithGrowZoneNumber(growZone.number)
    val customSortOrder = plantsListSortOrderCache.getOrAwait()
    emitSource(plantsGrowZoneLiveData.map { plantList -> 
        plantList.applySort(customSortOrder)
    })
}

现在，你再次运行app时，自定义排序列表效果将如下所示：
协程高级之结合Kotlin Flow与LiveData一起使用
自定义排序列表效果图

LiveData构造器允许我们进行异步计算，由于liveData是协程支持的。这里我们有一个挂起函数从数据库获取LiveData类型的植物列表，也调用一个挂起函数获取自定义排序列表。我们将这两个数据做合并后进行排序，最后返回结果，所有的操作都在构造器中完成。
小提示：
你可以使用emitSource()函数从LiveData发射多个数据，无论何时你想发射一个新的数据。注意每次调用emitSource()都将移除之前添加的数据源。

在LiveData开始被观察时，协程才开始执行。在协程成功执行完或者数据库和网络调用失败时，协程会被取消。
小提示：
如果挂起函数调用失败，整个代码块将被取消并且不会再次启动，从而避免泄漏。

接下来，我们将探索getPlantsWithGrowZone函数的几种变换操作。

7.修改liveData的值

我们将修改PlantRepository类，学习如何对LiveData做复杂的异步变换。作为一个先决条件，让我们创建排序算法的一个版本，可以在主线程中安全使用。我们可以使用witchContext来切换dispatcher。

在PlantRepository中添加如下代码：
PlantRepository.kt

@AnyThread
suspend fun List<Plant>.applyMainSafeSort(customSortOrder: List<String>) =
    withContext(defaultDispatcher) {
        this@applyMainSafeSort.applySort(customSortOrder)
    }

小提示:
协程使用withContext来切换dispatcher。默认情况下，Kotlin协程提供三种Dispatchers：Main，IO和Default。Main用于Android主线程，在更新界面刷新UI的场景使用，IO用于网络和数据库访问场景下使用，Default在CPU密集型任务场景下使用。

我们可以在LiveData构造器中使用新的main-safe排序。使用switchMap更新下面的代码块，每次接收到新的值时，switchMap都可以让你得到一个新的LiveData值。

PlantRepository.kt

fun getPlantsWithGrowZone(growZone: GrowZone) =
   plantDao.getPlantsWithGrowZoneNumber(growZone.number)
       .switchMap { plantList ->
           liveData {
               val customSortOrder = plantsListSortOrderCache.getOrAwait()
               emit(plantList.applyMainSafeSort(customSortOrder))
           }
       }

同之前的版本相比，一旦从网络接收到自定义排序列表，可以将它用于新的主线程安全的函数applyMainSafeSort。这个结果将传递给switchMap作为getPlantsWithGrowZone新的结果。

同上面plants的LiveData类似，协程在它被观察时开始执行，如果数据库或者网络调用失败，协程将被终止或者完成。所不同的是，在这里，从网络获取数据并缓存是安全的。

现在让我们看如何使用Flow来实现这段业务代码。

8. 引入Flow

小提示
Flow APIs是实验性的

我们将使用Flow实现相同的逻辑，该类位于kotlinx-coroutines Lib中。首先，让我们先来了解下flow。

Flow是异步版本的Sequence,一种类型的集合，所有的数据都是懒加载产生的。同sequence类似，flow按需产生数据，无论在何时我们需要数据时，并且flows包含的数据没有限制。

那么，为什么Kotlin要引入Flow类型呢，和常规的sequence有什么区别呢？答案就是神奇的async。Flow包含了协程的全部支持。那就意味着你可以使用协程来构造、转换，和消费一个Flow。你也可以控制并发，那就意味着你可以协调地执行多个有Flow声明的协程。

这开启了许多令人兴奋的肯能性。

小提示
Flow是一个异步的序列值,Flow一次产生一个值（而不是一次产生所有），这些值从异步操作产生，例如网络请求，数据库调用，或其他异步代码。它的API支持协程，所以你可以使用协程对一个flow进行变换。

Flow支持响应式的编程。如果你之前使用过RxJava，Flow提供了类似的功能。通过使用功能操作符转换流，可以简洁地表达应用程序的逻辑，这些操作符包括：map，flatMapLatest， combine等等。

Flow还支持大多数操作符上的挂起函数，这允许你在操作符内部执行顺序异步任务，例如map操作符。在一个flow内部使用挂起操作符，与完全响应式代码相比，代码更短、更易于阅读。

接下来，我们将学习使用这两种方式编程。

如何使用flow运行

要习惯于Flow如何按需（或懒加载）生成值，请看以下发射值的flow (1, 2, 3)， 在值产生前后进行打印。

fun makeFlow() = flow {
   println("sending first value")
   emit(1)
   println("first value collected, sending another value")
   emit(2)
   println("second value collected, sending a third value")
   emit(3)
   println("done")
}

scope.launch {
   makeFlow().collect { value ->
       println("got $value")
   }
   println("flow is completed")
}

如果你运行了上述代码，它将输出如下：

sending first value
got 1
first value collected, sending another value
got 2
second value collected, sending a third value
got 3
done
flow is completed

你可以看到集合lambda表达式和flow构造器之间代码是如何执行跳转的。每次flow构造器调用emit，它将suspends直到元素被完全处理。然后，当另外一个值从flow中请求，它将从它停止的地方resumes直到它再次调用emit。当flow构造完成，Flow将被取消，collect将恢复，调用协程打印flow is completed.

调用collect是非常重要的。Flow使用挂起操作符例如collect，而不是公开一个Iterator接口，所以它总是知道什么时候它被主动消费。更重要一点是，它知道什么时候调用者不可以再请求数据，从而可以清理资源。
小提示：
Flow是使用协程从头到尾构建的。通过使用协程的suspend和resume机制，他们可以同步生产者（flow）与消费者（collect）的执行。
如果你已经使用过了响应式编程，你一定熟悉背压的概念，Flow中也有实现，通过挂起一个协程来实现。
flow什么时候运行

上面的示例代码中，当collect操作符运行时，Flow也开始运行。调用flow构造器创建一个新的Flow 或者其它APIs没有引起任务的执行。挂起操作符collect在Flow中被称作terminal operator。也有一些其他挂起terminal operator，例如toList，first和single，这些操作符都位于kotlinx-coroutines包内，你也可以自己构造。

默认情况下Flow将执行：

  • 每次应用一个terminal operator时，内存已经不足了
  • 直到terminal operator被取消
  • 当最后一个值已被完全处理，另一个值被请求

小提示
这些规则是Flow的默认行为，并且可以为Flow分配内存，不会为每个terminal operator重启，并且通过Flow的内置或自定义转换独立于集合执行。

小Case
执行一个Flow被叫做collecting 一个flow。默认情况下，Flow将不做任何事情直到它被collected，这也意味着该Flow正在应用一个terminal operator。

myFlow.toList() // toList collects this flow and adds the values to a List

我们也说一个值是由terminal operator从Flow中收集的。

myFlow.collect { item -> println("$item has been collected") }

因为这些规则，Flow可以参与结构化并发，从一个Flow启动长时间运行的协程是安全的。Flow不会导致资源泄漏，当调用者被取消时，他们总是可以被清理干净，这有赖于协程取消规则

让我们修改上述flow，仅看前面两个元素，先使用take操作符，然后collect两次。

scope.launch {
   val repeatableFlow = makeFlow().take(2)  // we only care about the first two elements
   println("first collection")
   repeatableFlow.collect()
   println("collecting again")
   repeatableFlow.collect()
   println("second collection completed")
}

运行上述代码，将看到如下输出：

first collection
sending first value
first value collected, sending another value
collecting again
sending first value
first value collected, sending another value
second collection completed

flow的lambda表达式从顶部开始，在每次collect被调用时侯。这对于耗时的任务，例如网络请求，是非常重要的。另外，由于我们应用了take(2)操作符，这个flow将只会产生2个值。
在第二次调用emit之后，它将不会再恢复flow的lambda，因此"second value collected…"将不会被打印。
小提示
默认情况下，每次一个terminal operator被应用，Flow将从顶部重启。这对于Flow执行耗时的任务是很重要的，例如进行一个网络请求。

9. flow进行异步处理

Flow的懒加载属性有点像Sequence，但是它是如何做到异步的呢？让我们看一个例子，一个异步序列-观察数据的变更。

本例中，我们需要将数据库线程池中生成的数据与另一个线程(如主线程或UI线程)上的观察者进行协调。而且，由于随着数据的变化，我们会重复地发出结果，所以这种场景自然适合异步序列模式。

假设你的任务是为Flow编写Room集成。下面的代码中示例了Room中挂起查询的支持：

// This code is a simplified version of how Room implements flow
fun <T> createFlow(query: Query, tables: List<Tables>): Flow<T> = flow {
    val changeTracker = tableChangeTracker(tables)

    while(true) {
        emit(suspendQuery(query))
        changeTracker.suspendUntilChanged()
    }
}

这段代码依赖2个虚的挂起函数来生成一个Flow：

  • suspendQuery —— 一个主线程安全的函数，运行一个常规的Room挂起查询
  • suspendUntilChanged——一个挂起协程的函数直到其中一个表格内容变更

当collected时，flow将开始emits查询到的第一个值。一旦这个值被处理，这个flow将恢复并调用suspendUntilChanged函数，按上面的说法——挂起这个flow直到一个数据库表发生变更。在这点上，系统内没有任何事情发生直到某个数据库表内容发生变更并且这个flow将恢复。

当这个flow恢复时，它将发起另一个main-safe查询，并emits结果数据。这个过程将在一个无限中循环持续下去。

  • Flow与结构化并发

协程本身是轻量级的，但是它反复唤醒自己来执行数据库查询。这很容易导致泄漏。
尽管我们创建一个无限循环，Flow帮我们支持结构化并发。
通过flow消费数据的唯一途径就是使用terminal operator。因为所有的terminal operators是挂起函数，这些工作被绑定到相关作用域的生命周期。当作用域被取消时，flow将自动取消，这是根据协程取消规则.因此，尽管我们在我们的flow构造器中写入无限循环，我们安全地消费这个Flow而没有泄漏，由于结构化并发。

小Case
Flow支持结构化并发
因为一个flow只能通过terminal operators消耗数据，它也可以支持结构化并发。
当一个flow的消费者被取消，整个Flow都会被取消。由于结构化并发，从中间步骤泄漏协程是不可能的。

10.Flow配合Room使用

本节中，我们将学习如何将Flow与Room配合使用，并展示到用户UI。
本节中，有许多Flow的公共用法。Flow配合Room的操作有点类似LiveData，都是作为一个可观察的数据库查询。

更新Dao
打开文件PlantDao.kt， 添加2个查询返回Flow<List<Plant>>类型数据：

PlantDao.kt

@Query("SELECT * from plants ORDER BY name")
fun getPlantsFlow(): Flow<List<Plant>>

@Query("SELECT * from plants WHERE growZoneNumber = :growZoneNumber ORDER BY name")
fun getPlantsWithGrowZoneNumberFlow(growZoneNumber: Int): Flow<List<Plant>>

注意返回类型，和LiveData版本区分开来。我们将他们放在一起进行比较。

小Case

本节代码中，我将分别使用LiveData构造器和Flow对数据库做相同的变换。在一个生产app中，我们仅包含其一，但是将他们放一起进行比较是非常有用的，可以看看他们是如何工作的

通过指定一个Flow返回类型，Room将执行具有以下特征的查询：

  • Main-safetype - 一个Flow类型的查询总是运行在Room的执行器上，因此它们总是main-safe.不会在主线程上运行。
  • Observes changes - Room自动观察数据的变化并发射数据到flow中。
  • Async sequence – Flow在每次变化时会发射整个查询结果，不需要引入任何buffers。如果你返回一个Flow<List<T>>类型，这个flow将发射一个List<T>类型包含查询结果的所有行。它将像序列一样执行——一次发射一个查询结果并挂起直到需要发起下一个查询。
  • Cancellable –当搜集这些flows的作用域被取消，Room将取消对查询的观察。

这些结合起来，使Flow成为从UI层观察数据库的一种很好的返回类型。

更新repository

继续将新的返回值同UI连接起来，打开文件PlantRepository.kt，添加如下代码：
PlantRepository.kt

val plantsFlow: Flow<List<Plant>>
   get() = plantDao.getPlantsFlow()

fun getPlantsWithGrowZoneFlow(growZoneNumber: GrowZone): Flow<List<Plant>> {
   return plantDao.getPlantsWithGrowZoneNumberFlow(growZoneNumber.number)
}

现在，我们仅通过调用者传递Flow值。这和我们开始的时候一样，需要传递LiveData到ViewModel。

更新ViewModel
在PlantListViewModel.kt文件中，让我们暴露plantsFlow接口。
PlantListViewModel.kt

// add a new property to plantListViewModel

val plantsUsingFlow: LiveData<List<Plant>> = plantRepository.plantsFlow.asLiveData()

我们将保留LiveData版本的接口（val plants），将两者进行比较。

由于我们想要在UI层保留LiveData，我们将使用asLiveData扩展方法转换Flow为一个LiveData类型。同LiveData构造器一样，这为生成的LiveData添加了一个可配置的超时。这很好，因为它使我们在每次配置更改（例如设备旋转）时都不会重新启动查询。

小提示
asLiveData操作符转换一个Flow类型为LiveData，从而有超时配置。同liveData构造器一样，超时将帮助Flow重启。如果在超时之前，有另一个观察者观察该Flow，这个Flow将不被取消。

由于flow提供了main-safe并可以取消，你不用将其转换为LiveData，直接传递给UI。然而，本文中我们将坚持在UI层使用LiveData。

在ViewModel中，为init代码块添加缓存更新。这一步是可选的，但是如果你清空了缓存，并且没有添加这个调用，你将看不到任何数据。

PlantListViewModel.kt

init {
    clearGrowZoneNumber()  // keep this

    // fetch the full plant list
    launchDataLoad { plantRepository.tryUpdateRecentPlantsCache() }
}

更新Fragment

打开文件PlantListFragment.kt，改写subscribeUi函数使其返回LiveData类型的plantsUsingFlow

PlantListFragment.kt

private fun subscribeUi(adapter: PlantAdapter) {
   viewModel.plantsUsingFlow.observe(viewLifecycleOwner) { plants ->
       adapter.submitList(plants)
   }
}

运行这个app
再次运行这个app，你将使用Flow加载数据，看下效果！下一节，我们将学习Flow的转换操作。

11. 声明组合流

在该步骤中，将将对plantsFlow应用排序. 我们将使用flow的声明式API

什么是声明式API
声明式API是一种API风格，它是用来描述你的程序应该做什么而不是它该如何做。我们很熟悉的SQL就是声明式语言，它允许开发人员表示希望数据库查询什么，而不是如何执行查询。通过使用变换符，例如map，combine，或者mapLatest，我们可以在每个元素在流中以声明的方式流动时表示我们希望如何转换它。它甚至允许我们以声明的方式表示并发性，这确实可以简化代码。本节中，您将看到如何使用操作符告诉Flow启动两个协程，并以声明方式组合他们的结果。

打开PlantRepository.kt文件，定义一个私有flow customSortFlow：
PlantRepository.kt

private val customSortFlow = flow { emit(plantsListSortOrderCache.getOrAwait()) }

这个定义的Flow，当执行collected时，将调用getOrAwait并emit排序好的数据。
由于这个flow仅发射单个值，你也可以直接从getOrAwait函数构造它，getOrAwait将使用asFlow来构建。

// Create a flow that calls a single function
private val customSortFlow = plantsListSortOrderCache::getOrAwait.asFlow()

这段代码调用getOrAwait创建了一个新的Flow并并结果作为其第一个也是唯一的值发射。

它通过引用getOrAwait方法使用了::来实现，并且在结果的Function对象上调用了asFlow。
所以这些flows都做相同的事情，调用getOrAwait并在完成前发射结果。

以声明方式组合多个流
现在我们有两个flow，customSortFlow和plantsFlow，让我们用声明方式将他们组合！
添加一个combine操作符到plantsFlow：
PlantRepository.kt

private val customSortFlow = plantsListSortOrderCache::getOrAwait.asFlow()

val plantsFlow: Flow<List<Plant>>
   get() = plantDao.getPlantsFlow()
       // When the result of customSortFlow is available,
       // this will combine it with the latest value from
       // the flow above.  Thus, as long as both `plants`
       // and `sortOrder` are have an initial value (their
       // flow has emitted at least one value), any change 
       // to either `plants` or `sortOrder`  will call
       // `plants.applySort(sortOrder)`.
       .combine(customSortFlow) { plants, sortOrder ->
          plants.applySort(sortOrder) 
       }

combine操作符号将两个流进行组合。两个流将运行在各自的协程中，然后无论什么时候其中一个flow产生一个新值时，转换符将被调用用于处理发射出来的最近的值。

通过使用combine操作符号，我们可以将缓存的网络查询与数据库查询结合起来。它们将同时在不同的协程上运行。也就是说当Room开始发起一个网络请求时，Retrofit可以开始网络查询。然后，一旦两个流的结果都可用时，它将在我们对加载的plants排序地方调用combine lambda表达式。

小提示
combine操作符将为每个被combined的flow启动一个协程。这可以让你同时合并两个flows。它将以一种"公平"的方式将这些flow结合起来，也就是说他们都将得到产生一个值的机会(即使其中一个是由紧密循环产生的).
为了探究combine操作符如何工作，修改customSortFlow发射两次，并在onStart中有相当大延迟，如下所示：

// Create a flow that calls a single function
private val customSortFlow = suspend {() }.asFlow()
   .onStart {
       emit(listOf())
       delay(1500)
   }

当观察者在其他操作符之前监听时，onStart转换将发生，并且它可以发射占位符值。所以这里我们发射一个空列表，延迟调用getOrAwait 1500毫秒，然后继续原来的流程。如果您现在运行这个app，您将看到Room数据库查询立即返回，并与空列表相结合（这意味着它将按字母顺序排序）。大约1500毫秒后，它将应用自定义排序。
在继续使用codelab之前，请从customSortFlow中删除onStart转换。

小提示：
可以使用onStart在flow运行之前运行挂起的代码。它甚至可以向flow中发射额外的值，因此您可以使用它在网络请求flow上发射加载状态。

Flow与main-safety
Flow可以调用main-safe函数，就像我们在这里所做的那样，它将保证协程的正常的main-safety安全。Room与Retrofit给我们带来main-safety，我们不需要做任何其他事情来进行网络请求和数据库查询。

这个flow已使用了以下线程：

  • plantService.customPlantSortOrder运行在一个Retrofit线程（称作Call.enqueue）
  • getPlantsFlow将运行在Room的查询Executor
  • applySort将运行在搜集分发器(称作Dispatchers.Main)

因此，如果我们所做的只是在Retrofit中调用挂起函数并使用Room flows，那么我们就不需要将main-safety问题复杂化。

但是，随着数据集的增长，对applySort的调用可能会变得足够慢从而导致主线程阻塞。Flow提供了一个名为flowOn的声明式API来控制流在哪个线程上运行。

添加flowOn到plantsFlow中像下面这样：
PlantRepository.kt

private val customSortFlow = plantsListSortOrderCache::getOrAwait.asFlow()

val plantsFlow: Flow<List<Plant>>
   get() = plantDao.getPlantsFlow()
       .combine(customSortFlow) { plants, sortOrder ->
          plants.applySort(sortOrder) 
       }
       .flowOn(defaultDispatcher)
       .conflate()

调用flowOn对代码如何执行有重要影响：

  1. 在defaultDispatcher上启动一个新的协程（在这种情况下是Dispatchers.Default），在调用flowOn之前运行并搜集flow。
  2. 引入一个buffer区，将新协程的结果发送给后面的调用。
  3. 在flowOn之后，将值从buffer发射到FLow中。在这种场景下，有点类似ViewModel中的asLiveData函数

这和withContext中切换dispatchers的工作非常类似，但是它在我们转换过程中引入一个buffer，从而改变了flow的工作方式。由flowOn启动的协程生产的数据的速度可以比调用者消费它们的速度快，默认情况下它会缓冲大量数据集。

在这种情况，我们计划将结果发送到UI，因此我们之关心最近的结果，这就是conflate操作符所做的——它修改flowOn的buffer以只存储最后的结果。如果前一个结果还未被读取，新的结果又到来了，它将被覆盖。

小提示
操作符flowOn将启动一个新的协程来搜集它上面的flow并引入一个buffer来写入结果。
你可以使用很多操作符来控制这个buffer，例如conflate，这个操作符仅存储buffer中最后产生的值。

注意
在对诸如Room 结果之类的大型对象使用flowOn时，必须注意缓冲区，因为使用大量内存缓冲结果是很容易发生的。

运行app
再次运行这个app，你现在使用Flow加载数据并应用自定义排序。由于我们还没有实现switchMap操作符号的功能，这个过滤选型还没有做任何事情。
下一步，我们将使用flow另一种方式提供main safety.

12. 在2个flows之间切换

我完成此API的Flow版本，将打开PlantListViewModel.kt文件，我们将基于GrowZone的Flow之间切换，就像LiveData版本一样。

添加如下代码：
PlantsListViewModel.kt

private val growZoneChannel = ConflatedBroadcastChannel<GrowZone>()

val plantsUsingFlow: LiveData<List<Plant>> = growZoneChannel.asFlow()
    .flatMapLatest { growZone ->
        if (growZone == NoGrowZone) {
            plantRepository.plantsFlow
        } else {
            plantRepository.getPlantsWithGrowZoneFlow(growZone)
        }
    }.asLiveData()

注意：这个示例使用了@ExperimentalCoroutinesApis，而且在最终版本的Flow API中，可能会有一个更简洁的版本。

此方式显示如何将事件（grow zone更改）集成到flow中。它的功能和LiveData.switchMap一样——基于事件在两个数据源之间切换。

逐步完成代码
PlantListViewModel.kt

private val growZoneChannel = ConflatedBroadcastChannel<GrowZone>()

这段代码定义了一个新的ConflatedBroadcastChannel。这是一种特殊协程，基于值持有者，它只保存最后一个给定的值。它是一个线程安全的并发原语，因此你可以同时从多个线程对其进行写操作（以最后一个值为准）。
你还可以订阅以获取当前值的更新。总的来说，它的行为和LiveData类似—它只保存最后一个值，并允许你观察对它的更改。但是，与LiveData不同的是，你必须使用协程来读取多个线程上的值。
扩展阅读：
ConflatedBroadcastChannel通常是将事件插入FLow的好方法。它提供了一个并发原语（或底层工具），用于在多个协程之间传递值。
通过合并事件，我们只跟踪最近的事件。这通常是正确的做法，因为UI事件可能比处理更快，而且我们通常不关心中间值。

如果您确实需要在协程之间传递所有事件并且不希望合并，请考虑使用一个通道，该通道使用suspend函数提供BlockingQueue的语义。channelFlow 构造器可用于生成通道备份flows。
PlantListViewModel.kt

val plantsUsingFlow: LiveData<List<Plant>> = growZoneChannel.asFlow()

订阅ConflatedBroadcastChannel中的变更的最简单的方法之一是将其转换为flow。这将构建一个flow，在收集时，该流将订阅对ConflatedBroadcastChannel的变更，并在流上发送这些更改。它不会添加任何额外的buffer，因此如果流的收集器比写入`growZoneChannel的速度慢，它将跳过任何结果，只发射最近的结果。

这也很好，因为取消频道订阅将发生在流取消时。
扩展阅读：
asFlow是在ConflatedBroadcastChannel上的扩展函数，可以将一个ConflatedBroadcastChannel转换成Flow，这个Flow将与ConflatedBroadcastChannel有相同合并行为的流。

这是订阅ConflatedBroadcastChannel中变更的一种简单方法。

PlantListViewModel.kt

   .flatMapLatest { growZone ->
-
-

这与LiveData中switchMap完全相同。每当growtzonechannel更改其值时，将应用此lambda表达式，并且必须返回一个Flow。然后，返回的Flow将用作所有下游操作符的Flow。

基本上，这允许我们根据growZone的值在不同的flows之间切换。

扩展阅读：
Flow的flatMapLatest扩展函数允许你在多个flows之间切换。

PlantListViewModel.kt

if (growZone == NoGrowZone) {
    plantRepository.plantsFlow
} else {
    plantRepository.getPlantsWithGrowZoneFlow(growZone)
}

在flatMapLatest内，我们基于growZone切换。这段代码和LiveData.switchMap版本很像，一点不同的是它返回是Flows而不是LiveDatas。

PlantListViewModel.kt

   }.asLiveData()

最后，我们转换Flow为LiveData，由于我们的Fragment想要我们在ViewModel中暴露LiveData类型数据。

扩展阅读：
asLiveData操作符将转换Flow为LiveData，并有一个可配置的超时。
同liveData构造器一样，超时将使流在旋转过程中保持活动状态，这样collection就不会重启。

发送一个值到一个通道
为了让通道知道过滤器的更改，我们可以调用offer。这是一个常规（非挂起）函数。这是向协程通知事件的简单方式，就像我们正在做的一样。

在ViewModel中，在setGrowZoneNumber和clearGrowZoneNumber中调用offer的代码如下所示：
PlantListViewModel.kt

fun setGrowZoneNumber(num: Int) {
    growZone.value = GrowZone(num)
    growZoneChannel.offer(GrowZone(num))

    launchDataLoad {
        plantRepository.tryUpdateRecentPlantsForGrowZoneCache(GrowZone(num))
    }
}

fun clearGrowZoneNumber() {
    growZone.value = NoGrowZone
    growZoneChannel.offer(NoGrowZone)

    launchDataLoad {
        plantRepository.tryUpdateRecentPlantsCache()
    }
}

再次运行app
再次运行你的app，LiveData版本和Flow版本的过滤器都可以工作了。
下一步，我们对getPlantsWithGrowZoneFlow进行自定义排序。

13. 混合风格的flow

Flow的一个最令人兴奋的功能对挂起函数的支持。flow生成器和几乎每个转换都公开了一个可以调用任何挂起函数的挂起操作符。因此，网络和数据库调用以及编排多个异步操作的main-safety可以通过从flow中调用常规挂起函数来实现。

实际上，这允许您自然地将声明性转换与命令式代码混合。正如您将在本例中看到的，在常规map操作符中，您可以编排多个异步操作，而无需应用任何额外的转换。在很多地方，这会导致代码比完全声明性方法简单得多。

扩展阅读：
如果您已经广泛地使用了RxJava这样的库，这是Flow提供的主要区别之一。

开始使用Flow时，请仔细考虑如何使用挂起转换来简化代码。在许多情况下，通过在map、onStart和onCompletion等操作符中依赖挂起操作，可以自然地表达异步代码。

来自Rx的熟悉操作符，如combine、mapLatest、flatMapLatest、flatMapMerge和flatMapMerge，最好用于编排FLow中的并发操作。

使用挂起函数编排异步任务

我了结束我们对Flow的探索，我们将使用suspend操作符应用自定义排序。

打开PlantRepository.kt文件，为getPlantsWithGrowZoneNumber添加一个map转换.

PlantRepository.kt

fun getPlantsWithGrowZoneFlow(growZone: GrowZone): Flow<List<Plant>> {
   return plantDao.getPlantsWithGrowZoneNumberFlow(growZone.number)
       .map { plantList ->
           val sortOrderFromNetwork = plantsListSortOrderCache.getOrAwait()
           val nextValue = plantList.applyMainSafeSort(sortOrderFromNetwork)
           nextValue
       }
}

通过依赖常规的挂起函数来处理异步工作，这个map操作符是main-safe的，即使它结合了两个异步操作。

当来自数据库的每个结果返回时，我们将获得缓存的排序顺序——如果它还没有准备好，它将等待异步网络请求。一旦我们有了排序顺序，就可以安全地调用applyMainSafeSort，它将在默认调度器上运行排序。

通过将主安全问题推迟到常规的挂起函数，该代码现在完全是main-safe。它比plantsFlow中实现的相同转换要简单得多。

扩展阅读：
在Flow中，map及其他操作符提供了挂起lambda。
通过使用协程的挂起与恢复机制，你通常可以很轻松地编排顺序异步调用，而无需使用声明性转换。

上一篇：CF235C-Cyclical Quest【SAM】


下一篇：最为详尽的关于PP中替代及取代功能的详解