协程与协程间通信：Channel的跨协程通信、Flow的共享与重放、select表达式处理多路复用、协程间数据同步策略

协程之间的通信，说白了就是「怎么让两个协程好好说话」。

我刚开始用协程时，总觉得每个协程都是独立的孤岛。后来发现，真正复杂的业务场景，往往需要多个协程协同工作。比如一个协程在下载数据，另一个在更新UI，第三个在缓存结果——它们之间怎么传递消息？怎么避免数据竞争？

嗯，这就是我们今天要聊的核心问题。

Channel：协程之间的管道

Channel 的概念其实很简单——它就是一个队列。一个协程往里面放数据，另一个协程从里面取数据。跟 BlockingQueue 很像，但它是挂起的，不是阻塞的。

核心区别：BlockingQueue 会让线程卡住，Channel 只会让协程挂起。线程不释放，协程可以继续干别的。

来看个最简单的例子：

val channel = Channel<Int>()

// 生产者协程
launch {
    for (i in 1..5) {
        delay(100)
        channel.send(i)  // 挂起，直到有人接收
    }
    channel.close()  // 记得关闭
}

// 消费者协程
launch {
    for (value in channel) {  // 自动迭代，直到channel关闭
        println("收到: $value")
    }
}

我在项目中遇到过一个问题：生产者生产速度远快于消费者，结果内存暴涨。为什么？因为 Channel 默认是 Rendezvous 模式——没有缓冲区，send 必须等 receive。但如果消费者处理慢，send 就一直挂起，不会堆积。

等等，那内存暴涨是怎么回事？

后来发现是用了 Channel(Channel.UNLIMITED)。无限制缓冲区，生产者疯狂塞数据，消费者来不及处理，内存就炸了。

Channel 的四种类型

类型 缓冲区大小 行为特点 适用场景
Rendezvous 0 send 和 receive 必须同时发生 一对一精确同步
Buffered 指定大小 缓冲区满时 send 挂起 生产消费速度不匹配
UNLIMITED 无限制 永不挂起，但可能 OOM 数据量可控的场景
CONFLATED 1 新数据覆盖旧数据 只关心最新值

我的建议：大多数情况下用 Buffered 就够了。UNLIMITED 慎用，除非你确定数据量不会爆炸。CONFLATED 适合「只关心最新状态」的场景，比如位置更新。

Flow 的共享与重放

Channel 是点对点的，而 Flow 更像是广播。一个 Flow 可以有多个订阅者，每个订阅者独立接收数据。

但这里有个坑：普通的 Flow 每次 collect 都会重新执行上游代码。比如：

val flow = flow {
    emit("第一次执行")
    emit("第二次执行")
}

// 第一个订阅者
flow.collect { println(it) }  // 输出两次
// 第二个订阅者
flow.collect { println(it) }  // 又输出两次

每次 collect 都重新执行，这在某些场景下是浪费。比如网络请求的结果，你希望多个订阅者共享同一个结果，而不是每个订阅者都重新请求一次。

shareIn 与 stateIn

这时候就需要 共享 了。Kotlin 提供了两个操作符：

  • shareIn：共享 Flow，多个订阅者共用同一个上游
  • stateIn：共享并缓存最新值，类似 StateFlow
val sharedFlow = flow {
    emit("数据1")
    emit("数据2")
}.shareIn(
    scope = viewModelScope,
    started = SharingStarted.WhileSubscribed(),
    replay = 1  // 重放最近1个值
)

这里有个参数 started，它控制共享何时开始：

策略 行为
Eagerly 立即启动，不管有没有订阅者
Lazily 第一个订阅者到来时启动
WhileSubscribed() 有订阅者时启动，全部取消后停止

我曾经踩过的坑：用 Eagerly 启动一个网络请求的共享 Flow，结果页面还没打开，请求就发出了。用户切到后台再回来，请求已经过期了。后来改成 WhileSubscribed()，配合超时参数，才解决问题。

replay 的作用

replay 参数决定了新订阅者能收到多少个历史值。比如 replay = 2，新订阅者一进来就能收到最近的两个值。

这个在 UI 场景下特别有用。比如用户旋转屏幕，Activity 重建，新的订阅者需要立即拿到当前状态，而不是等下一次数据更新。

select 表达式：多路复用

有时候你需要同时监听多个数据源，哪个先来就处理哪个。这就是 多路复用。

Kotlin 的 select 表达式就是干这个的：

suspend fun selectFromChannels(
    channel1: Channel<String>,
    channel2: Channel<String>
): String = select {
    channel1.onReceive { value ->
        "来自channel1: $value"
    }
    channel2.onReceive { value ->
        "来自channel2: $value"
    }
}

哪个 Channel 先有数据，select 就返回哪个结果。这比用两个协程分别监听要优雅得多。

select 的常见用法：

  • 多个 Channel 的竞争接收
  • Channel 与 Deferred 的混合等待
  • 超时控制（配合 onTimeout）

来看一个更实际的例子——同时等待用户输入和超时：

val userInput = Channel<String>()
val timeout = 5000L

select<Unit> {
    userInput.onReceive { input ->
        println("用户输入: $input")
    }
    onTimeout(timeout) {
        println("超时了，使用默认值")
    }
}

这个模式在搜索框自动补全、表单验证等场景下非常实用。我做过一个搜索功能，用户每输入一个字就触发搜索，但网络慢的时候会堆积请求。用 select 配合 debounce，完美解决。

协程间数据同步策略

多个协程访问共享数据，必然涉及同步问题。Kotlin 协程提供了几种方案：

1. 互斥锁：Mutex

Mutex 是协程版的锁。跟 synchronized 不同，它不会阻塞线程，只会挂起协程。

val mutex = Mutex()
var counter = 0

suspend fun increment() {
    mutex.withLock {
        counter++
    }
}

用 withLock 确保同一时间只有一个协程执行临界区代码。我习惯用它来保护「写多读少」的共享状态。

2. 原子操作：Atomic 类

对于简单的计数器、标志位，用 AtomicInteger、AtomicBoolean 就够了。它们比 Mutex 轻量，因为不需要挂起。

val counter = AtomicInteger(0)

fun increment() {
    counter.incrementAndGet()  // 不需要挂起
}

我的选择原则：如果只是简单的数值增减，用 Atomic。如果涉及多个变量的组合操作，用 Mutex。别为了炫技用 Mutex 保护一个 int，那是杀鸡用牛刀。

3. 单线程约束

有时候最简单的方案就是「把所有共享状态的访问都放到同一个协程里」。比如用 actor 模式：

val actor = CoroutineScope(Dispatchers.Default).actor<Action> {
    var state = 0
    for (action in channel) {
        when (action) {
            is Action.Increment -> state++
            is Action.Get -> action.callback(state)
        }
    }
}

所有对 state 的修改都通过 actor 的 channel 发送消息，天然避免了并发问题。这个模式在复杂的业务逻辑中特别好用。

4. 不可变数据 + 快照

如果你能接受「每次修改都创建新对象」，那就用不可变数据。配合 StateFlow，每次更新都 emit 一个新对象：

data class UiState(
    val list: List<Item> = emptyList(),
    val isLoading: Boolean = false
)

private val _state = MutableStateFlow(UiState())
val state: StateFlow<UiState> = _state.asStateFlow()

fun updateList(newList: List<Item>) {
    _state.update { it.copy(list = newList) }
}

这种方式没有锁，没有原子操作，但完全线程安全。因为 StateFlow 的 update 方法内部是原子的。

注意：不可变数据虽然安全，但如果数据量很大，频繁 copy 会有性能开销。我一般在 UI 层用这个模式，数据量通常不大。如果是大量数据的批量更新，还是用 Mutex 更合适。

知识体系总览

下面这张图总结了协程间通信的核心概念和它们之间的关系：

协程间通信知识体系 Channel Flow select Rendezvous Buffered UNLIMITED CONFLATED shareIn stateIn replay WhileSubscribed onReceive onTimeout onAwait 协程间数据同步策略 Mutex 互斥锁 Atomic 原子类 单线程约束 不可变数据

从图中可以看到，Channel 适合点对点通信，Flow 适合一对多广播，select 则是在多个数据源中做选择。而底层的同步策略，是保证数据一致性的基石。

在实际项目中，我通常这样组合使用：

  • 用 Channel 做「事件驱动」的通信，比如点击事件、网络状态变化
  • 用 StateFlow 做「状态持有」，比如 UI 状态、用户信息
  • 用 select 做「竞争处理」，比如多个数据源谁先到就用谁
  • 用 Mutex 保护「写多读少」的共享资源

记住一点：没有银弹。每种方案都有它的适用场景。你想想看，如果所有地方都用 Channel，那 Flow 的共享特性就浪费了。反过来，如果所有状态都用 StateFlow，那事件的一次性消费就不好处理。

嗯，理解了这些，你在协程间通信这块就算真正入门了。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →