ZH
FrançaisDeutschEnglishEspañolPortuguêsTiếng Việt한국어简体中文日本語
首页 / 编程语言 / Kotlin

Kotlin Flow: 异步数据流

Cris The Dev · 2026/10/13 · 9 分钟阅读 · 103

1. 总览

一个 suspend 函数只返回一个值。当你需要接收多个值随时间变化时，例如新订单、支付状态或股票价格，你需要一个数据流。Kotlin Flow 正是为此而生。

本文解释什么是 Flow 以及如何使用操作符转换数据。之后你将学会使用 flowOn 切换线程、处理错误、使用 StateFlow 和 SharedFlow，以及为 Flow 编写测试。

2. 安装

Flow 位于 kotlinx-coroutines 库中。本文假设你的 Maven 项目已配置 kotlin-maven-plugin。你只需添加以下依赖项：

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

如果你使用 Spring Boot 3.x，coroutines 版本已通过 BOM 管理，所以你可以省略 <version> 标签。

3. 第一个 Flow

3.1. 创建和收集 Flow

创建 Flow 的最简单方法是使用 flow { } 构造器并为每个值调用 emit()：

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("开始发送订单")
    for (id in 1..3) {
        delay(100)                      // 模拟等待来自系统的数据
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("已创建 flow")
    orders.collect { println("接收: $it") }
}

解释：

  • orderStream() 不是 suspend 函数。它只返回对数据流的描述。
  • 在 flow { } 内，你可以调用 suspend 函数如 delay()。
  • collect 是一个 terminal operator。只有调用它时，flow { } 内的代码才会运行。

结果：

已创建 flow
开始发送订单
接收: Order(id=1, amount=50000)
接收: Order(id=2, amount=100000)
接收: Order(id=3, amount=150000)

“已创建 flow”这一行出现在“开始发送订单”之前。这证明创建 Flow 不会触发任何操作。

3.2. Flow 是“cold”的

flow { } 创建的 Flow 是一个 cold stream。每次 collect 都会从头开始重新运行整个代码块：

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}
开始发送订单
开始发送订单

注意： 如果 flow { } 块调用 API 或写入数据库，每次 collect 都会重新调用。这是许多“为什么 API 被调用两次”bug 的根源。如果想与多个收集器共享一个源，请参阅第 6 节。

4. 转换操作符

4.1. Intermediate operator

Flow 拥有一组熟悉的操作符，如集合中的 map、filter、take、onEach 等。这些操作符是 intermediate。它们返回一个新的 Flow，也是 lazy 的，就像原始 Flow 一样。

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "订单 #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}
开始发送订单
订单 #2: 100000 VND
订单 #3: 150000 VND

每个元素通过整个 filter → map → collect 链，然后才发送下一个元素。Flow 默认在同一个 coroutine 中顺序运行。

与集合的 map 不同，Flow 的 map 中的 lambda 是 suspend 的。因此你可以直接在其中调用异步函数：

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "客户 $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. 其他 Terminal operator

除了 collect，你还可以使用其他 terminal operator 来直接获取结果：

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("订单数: ${all.size}, 首个大订单: #${firstBig.id}, 收入: $revenue")
}
...
订单数: 3, 首个大订单: #2, 收入: 300000

first { } 在找到匹配元素时立即取消 Flow。因此订单 #3 在那次 collect 中永远不会被发送。

5. 使用 flowOn 切换线程

默认情况下，flow { } 内的代码在调用 collect 的 coroutine 的 context 中运行。要将数据发送部分推送到另一个线程，例如读取文件或数据库，你可以使用 flowOn：

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("在 ${Thread.currentThread().name} 上读取数据")
    listOf("Alice", "Bob").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("在 ${Thread.currentThread().name} 上处理 ${it.name}")
    }
}
在 DefaultDispatcher-worker-1 上读取数据
在 main 上处理 Alice
在 main 上处理 Bob

flowOn 只影响 upstream 部分，即它之前的操作符。collect 部分仍在调用者的 context 中运行。这个原则称为 context preservation。

注意： 不要使用 withContext 来包裹 flow { } 内的 emit()。下面的代码会抛出 IllegalStateException: Flow invariant is violated：

flow {
    withContext(Dispatchers.IO) { emit(1) } // 错误
}

始终使用 flowOn。如果你真的需要从多个不同的 coroutine 发送，请使用 channelFlow { }。

6. 错误处理

6.1. catch 和 onCompletion

Flow 中的异常向下传播到 collect，就像普通代码一样。catch 操作符允许你拦截错误并发送替代值：

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("支付网关超时")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("失败: ${e.message}") }
        .onCompletion { cause -> println("完成，原因=$cause") }
        .collect { println(it) }
}
PENDING
失败: 支付网关超时
完成，原因=null

解释：

  • catch 只捕获来自 upstream 的错误。在 collect 的 lambda 中抛出的异常不会被它捕获。
  • onCompletion 位于 catch 之后。错误已在前面的步骤中处理，所以 cause 是 null。如果交换两个操作符，你会看到 cause 包含异常。

6.2. retry

对于临时错误如网络中断，你可以使用 retry。这个操作符从头开始重新 collect upstream Flow：

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("连接丢失")
        emit("在 $attempts 次尝试后成功")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}
在 3 次尝试后成功

Predicate { e is IOException } 帮助你只重试网络错误。业务错误如 IllegalArgumentException 不会被重试。在 predicate 中，你可以调用 delay() 来创建 backoff。

7. StateFlow 和 SharedFlow：hot stream

flow { } 是 cold 的。但很多时候你需要一个独立于收集器的数据源，例如购物车状态或系统事件。这时你需要一个 hot stream。

7.1. StateFlow

StateFlow 总是保持一个当前值。新的收集器会立即收到该值：

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("购物车总计: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}
购物车总计: 0
购物车总计: 100000
购物车总计: 350000

需要注意的几点：

  • 属性 _total 是 private 且可写。外部只能看到只读的 StateFlow。这是封装状态的标准模式。
  • update { } 是原子更新，当多个线程同时调用 add 时是安全的。
  • StateFlow 永远不会结束。因此你必须 cancel job，否则 runBlocking 会永久等待。
  • StateFlow 跳过与当前值相同的值。它也会 conflate：慢的收集器只能看到最新值，看不到中间的步骤。

7.2. SharedFlow

SharedFlow 适合用于事件，例如“订单已支付”。这里每次发送都有意义，没有“当前值”的概念：

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("订单 #$orderId 已支付")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // 还没有人监听 -> 丢失

    val job = launch { payments.events.collect { println("邮件: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}
邮件: 订单 #1 已支付
邮件: 订单 #2 已支付

使用默认配置（replay = 0），在没有订阅者时发送的事件会丢失。如果收集器迟到需要接收最近的几个事件，请使用 MutableSharedFlow<String>(replay = 1)。

你也可以使用 stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) 或 shareIn(...) 将 cold Flow 转换为 hot Flow。这样多个收集器会共享一次 upstream 运行。

8. 测试 Flow

kotlinx-coroutines-test 库提供 runTest。这个函数使用虚拟时间跳过 delay，所以测试运行几乎是瞬间的：

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `只保留 100000 及以上的订单`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `支付错误转换为 FAILED 状态`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `购物车累加值`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

对于有限的 Flow，toList() 是检查所有发出值的最简单方法。对于 StateFlow，你直接读取 .value。不要在 StateFlow 上调用 toList()，因为这个 Flow 永不结束，测试会挂起。

9. 常见错误

  • collect 之后的代码不运行。 collect 是一个 suspend 函数，只有在 Flow 结束时才返回。对于 StateFlow/SharedFlow，这永远不会发生。将每个无限的 collect 放在单独的 launch 中。
  • 意外多次调用 API。 cold Flow 的每个收集器都会重新运行 upstream。当需要共享时，使用 shareIn/stateIn。
  • 在 emit 周围使用 withContext。 改用 flowOn 或 channelFlow。
  • 认为 catch 捕获所有错误。 这个操作符只捕获 upstream 错误。如果想在处理部分捕获错误，将逻辑移到 catch 前的 onEach { }，然后调用无参数的 collect()。
  • 对事件使用 StateFlow。 两个连续相同的事件会被合并为一个。对事件使用 SharedFlow。

注意： 当你只有一个结果时，不应该使用 Flow。一个简单的 suspend fun findOrder(id: Int): Order 函数比只发送一个元素的 Flow<Order> 更简单易读。

10. 总结

在本文中，你看到了 Flow 是异步数据流，是 cold 和 lazy 的。你学会了如何使用 map/filter 转换数据，使用 flowOn 安全地切换线程，以及使用 catch、retry、onCompletion 处理错误。对于共享状态和事件的需求，StateFlow 和 SharedFlow 是两个主要工具。runTest 帮助你快速稳定地测试它们。

从这里，你可以尝试高级操作符如 combine、flatMapLatest、debounce 或 buffer，以构建更复杂的数据管道。

kotlincoroutinesflow异步stateflow
作者的应用HoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同主题文章

Kotlin · 2026/10/26 · 9 分钟阅读 · 44

Kotlin 中的高阶函数与 lambda

学习如何声明函数类型、传递和返回 lambda、使用函数引用、闭包、lambda with receiver 以及 Kotlin 中的 inline。