Android 上的 Kotlin Flow

在协程中，Flow（数据流）是一种可以按顺序发出多个值的类型，这与仅返回单个值的 挂起函数 不同。例如，您可以使用 Flow 从数据库接收实时更新。

Flow 构建于协程之上，可以提供多个值。从概念上讲，Flow 是一个可以异步计算的 数据流。发出的值必须具有相同的类型。例如，Flow<Int> 就是一个发出整数值的流。

Flow 与生成值序列的 Iterator 非常相似，但它使用挂起函数异步生成和消费值。这意味着，例如，Flow 可以安全地发起网络请求来生成下一个值，而不会阻塞主线程。

数据流涉及三个实体

  • 生产者 (Producer)：负责生产添加到流中的数据。得益于协程，Flow 也可以异步生产数据。
  • 中介 (Intermediaries，可选)：可以修改流中发出的每个值或修改流本身。
  • 消费者 (Consumer)：负责消费流中的值。

entities involved in streams of data; consumer, optional intermediaries, and producer
图 1. 数据流涉及的实体：消费者、可选的中介和生产者。

在 Android 中，代码库 (Repository) 通常是 UI 数据的生产者，而用户界面 (UI) 是最终显示数据的消费者。有时，UI 层也是用户输入事件的生产者，由层级结构中的其他层进行消费。生产者和消费者之间的层通常充当中介，用于修改数据流以适应下一层的需求。

创建 Flow

要创建 Flow，请使用 flow 构建器 API。flow 构建函数会创建一个新的流，您可以在其中使用 emit 函数手动向数据流发出新值。

在下面的示例中，数据源会自动以固定的时间间隔获取最新新闻。由于挂起函数无法返回多个连续的值，因此数据源会创建并返回一个 Flow 来满足此需求。在这种情况下，数据源充当生产者。

class NewsRemoteDataSource(
    private val newsApi: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        while(true) {
            val latestNews = newsApi.fetchLatestNews()
            emit(latestNews) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}

// Interface that provides a way to make network requests with suspend functions
interface NewsApi {
    suspend fun fetchLatestNews(): List<ArticleHeadline>
}

flow 构建器在协程内执行。因此，它受益于相同的异步 API，但需要遵守一些限制

  • Flow 是 顺序的。由于生产者位于协程中，当调用挂起函数时，生产者会挂起，直到挂起函数返回。在示例中，生产者会挂起，直到 fetchLatestNews 网络请求完成。只有这样，结果才会发送到流中。
  • 使用 flow 构建器时，生产者不能在不同的 CoroutineContext 中 emit 值。因此，不要通过创建新协程或使用 withContext 代码块在不同的 CoroutineContext 中调用 emit。在这些情况下，您可以使用其他 Flow 构建器，例如 callbackFlow。

修改流

中介可以使用 中间运算符 (intermediate operators) 来修改数据流，而无需消费这些值。这些运算符是函数，当应用于数据流时，它们会设置一系列操作链，这些操作只有在未来消费这些值时才会执行。如需了解有关中间运算符的更多信息，请参阅 Flow 参考文档。

在下面的示例中，存储库层使用中间运算符 map 来转换要在 View 上显示的数据

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            // Intermediate operation to filter the list of favorite topics
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            // Intermediate operation to save the latest news in the cache
            .onEach { news -> saveInCache(news) }
}

中间运算符可以一个接一个地应用，形成一个操作链，当有项目发送到 Flow 中时，这些操作会被延迟执行。请注意，仅仅将中间运算符应用于流并不会开始 Flow 的收集。

收集 Flow

使用 末端运算符 (terminal operator) 来触发 Flow 开始监听值。要获取流中发出的所有值，请使用 collect。您可以在 Flow 官方文档 中详细了解末端运算符。

由于 collect 是一个挂起函数，因此它需要在协程内执行。它接受一个 lambda 作为参数，该 lambda 会在每次有新值时被调用。由于它是一个挂起函数，调用 collect 的协程可能会挂起，直到 Flow 关闭。

继续上面的示例，这是一个从存储库层消费数据的 ViewModel 的简单实现

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            newsRepository.favoriteLatestNews.collect { favoriteNews ->
                // Update View with the latest favorite news
            }
        }
    }
}

收集 Flow 会触发生产者，它会刷新最新新闻并以固定间隔发送网络请求的结果。由于生产者通过 while(true) 循环保持活跃，因此当 ViewModel 被清除且 viewModelScope 被取消时，数据流将关闭。

Flow 收集可能会因以下原因停止

  • 进行收集的协程被取消，如上一个示例所示。这也停止了底层的生产者。
  • 生产者停止发出项目。在这种情况下，数据流关闭，调用 collect 的协程恢复执行。

除非指定了其他中间运算符，否则 Flow 是 冷 (cold) 且 惰性 (lazy) 的。这意味着每次在 Flow 上调用末端运算符时，都会执行生产者代码。在前面的示例中，拥有多个 Flow 收集器会导致数据源在不同的固定间隔内多次获取最新新闻。若要在多个消费者同时收集时优化并共享 Flow，请使用 shareIn 运算符。

捕获意外异常

生产者的实现可能来自第三方库。这意味着它可能会抛出意外异常。要处理这些异常，请使用 catch 中间运算符。

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Intermediate catch operator. If an exception is thrown,
                // catch and update the UI
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                    // Update View with the latest favorite news
                }
        }
    }
}

在前面的示例中，当发生异常时，由于没有收到新项目，collect lambda 不会被调用。

catch 也可以向 Flow emit 项目。示例中的存储库层可以改用 emit 缓存值

class NewsRepository(...) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // If an error happens, emit the last cached values
            .catch { exception -> emit(lastCachedNews()) }
}

在此示例中，当发生异常时，collect lambda 会被调用，因为由于异常，一个新项目已被发送到流中。

在不同的 CoroutineContext 中执行

默认情况下，flow 构建器的生产者在收集它的协程的 CoroutineContext 中执行，并且如前所述，它不能在不同的 CoroutineContext 中 emit 值。这种行为在某些情况下可能并不理想。例如，在本主题使用的示例中，存储库层不应执行 viewModelScope 所使用的 Dispatchers.Main 上的操作。

要更改 Flow 的 CoroutineContext，请使用中间运算符 flowOn。flowOn 会更改 上游 Flow 的 CoroutineContext，这意味着生产者以及 flowOn 之前（或上方） 应用的任何中间运算符都会受到影响。下游 Flow（flowOn 之后 的中间运算符以及消费者）不受影响，并在用于 collect Flow 的 CoroutineContext 上执行。如果存在多个 flowOn 运算符，每一个都会从其当前位置更改上游。

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // Executes on the default dispatcher
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // Executes on the default dispatcher
                saveInCache(news)
            }
            // flowOn affects the upstream flow ↑
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ is not affected
            .catch { exception -> // Executes in the consumer's context
                emit(lastCachedNews())
            }
}

使用此代码，onEach 和 map 运算符使用 defaultDispatcher，而 catch 运算符和消费者则在 viewModelScope 所使用的 Dispatchers.Main 上执行。

由于数据源层正在执行 I/O 工作，您应该使用针对 I/O 操作进行了优化的调度程序

class NewsRemoteDataSource(
    ...,
    private val ioDispatcher: CoroutineDispatcher
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        ...
    }
        .flowOn(ioDispatcher)
}

Jetpack 库中的 Flow

Flow 已集成到许多 Jetpack 库中，在 Android 第三方库中也很受欢迎。Flow 非常适合实时数据更新和无限数据流。

您可以使用 Flow 与 Room 来获取数据库更改的通知。使用 数据访问对象 (DAO) 时，返回 Flow 类型以获取实时更新。

@Dao
abstract class ExampleDao {
    @Query("SELECT * FROM Example")
    abstract fun getExamples(): Flow<List<Example>>
}

每当 Example 表发生更改时，就会发出一个包含数据库中新项目的新列表。

将基于回调的 API 转换为 Flow

callbackFlow 是一个 flow 构建器，允许您将基于回调的 API 转换为 Flow。例如，Firebase Firestore Android API 使用回调。

要将这些 API 转换为 Flow 并监听 Firestore 数据库更新，您可以使用以下代码

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                trySend(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

与 flow 构建器不同，callbackFlow 允许使用 send 函数从不同的 CoroutineContext 发出值，或者使用 trySend 函数在协程外发出值。

在内部，callbackFlow 使用 通道 (channel)，在概念上它与阻塞 队列 (queue) 非常相似。通道配置有 容量 (capacity)，即可以缓冲的最大元素数量。callbackFlow 中创建的通道默认容量为 64 个元素。当您尝试向已满的通道添加新元素时，send 会挂起生产者，直到有空间容纳新元素为止，而 trySend 则不会将元素添加到通道并立即返回 false。

trySend 仅在不违反其容量限制的情况下立即将指定元素添加到通道，然后返回成功结果。

更多 Flow 资源