Kotlin协程SharedFlow结合gRPC服务端流时数据丢失问题求助

1. SharedFlow默认缓存策略缺陷 #

你使用shareIn(coroutineScope, SharingStarted.Eagerly)时，默认参数replay=0、extraBufferCapacity=0：

• 当SharedFlow没有活跃订阅者时，上游gRPC流发射的数据会直接被丢弃——既没有缓存空间存储，也没有订阅者接收。
• 若外部collector在gRPC流已开始推送数据后才订阅，会错过所有之前发射的数据。

2. gRPC协程流的冷流特性 #

gRPC Kotlin的服务端流属于冷流：只有当第一个collect操作触发时，才会发起gRPC请求并开始接收服务端数据。在你的初始版本中，SharingStarted.Eagerly会立即启动上游流的收集，但此时无活跃订阅者，导致初始阶段的数据直接丢失；后续外部订阅只能收到订阅后的新数据。

3. 内部collector的关键作用 #

修改后添加的内部collector，成为了SharedFlow的第一个活跃订阅者：

• 它会立即触发gRPC请求，确保数据从一开始就被接收。
• 由于该collector持续活跃，SharedFlow会将所有上游数据转发给它，同时后续外部订阅的collector会从订阅时刻开始接收实时数据，不会出现丢失（此时有活跃订阅者，上游数据会同步分发给所有订阅者）。

根据需求可选择以下方案：

方案1：配置SharedFlow缓存策略 #

如果需要让后续订阅者获取历史数据，调整shareIn的缓存参数：

fun createStream(): SharedFlow<T> {
  val stub = ServiceGrpcKt.Stub(channel)
  return stub.stream(request, header)
    .shareIn(
      coroutineScope,
      SharingStarted.Eagerly,
      replay = 10, // 保留最近10条历史数据
      extraBufferCapacity = 5 // 额外缓冲突发数据，避免丢失
    )
    .also { sharedFlow = it }
}

• replay：新订阅者订阅时会收到的历史数据条数。
• extraBufferCapacity：订阅者处理速度跟不上时的临时缓冲空间。

方案2：保持常驻活跃订阅者 #

如果只需要实时数据不丢失，可保留一个极简的内部订阅者：

fun createStream(): SharedFlow<T> {
  val stub = ServiceGrpcKt.Stub(channel)
  return stub.stream(request, header)
    .shareIn(coroutineScope, SharingStarted.Eagerly)
    .also { sharedFlow = it
      // 保持活跃订阅者，确保数据不被丢弃
      coroutineScope.launch { it.collect() }
    }
}

此方案确保SharedFlow始终有活跃订阅者，上游数据会被持续转发给所有后续订阅者。

方案3：调整流启动策略 #

若希望按需启动gRPC流（仅当有订阅者时才发起请求），改用WhileSubscribed策略：

fun createStream(): SharedFlow<T> {
  val stub = ServiceGrpcKt.Stub(channel)
  return stub.stream(request, header)
    .shareIn(
      coroutineScope,
      SharingStarted.WhileSubscribed(
        stopTimeoutMillis = 5000, // 最后一个订阅者取消后，延迟5秒停止上游流
        replayExpirationMillis = 0
      ),
      replay = 0
    )
    .also { sharedFlow = it }
}

这种策略避免了无订阅时的无效数据收集，适合按需使用流的场景。

内容的提问来源于stack exchange，提问作者이태훈