如何监控Kotlin Flow中flowOn及MutableSharedFlow的缓冲区大小？

目前kotlinx.coroutines官方没有提供公开API获取Flow内部缓冲区的当前大小，这类缓冲区默认作为内部实现细节存在，不支持外部直接访问。如果确实有监控需求，可根据场景选择以下两种方案：

方案1：反射获取（不推荐生产环境使用） #

该方案依赖kotlinx.coroutines的内部实现，版本迭代时可能会失效：

• 针对MutableSharedFlow：可通过反射获取其内部缓冲队列的实例，直接读取当前缓冲元素数量。目前的实现类中，缓冲数据存储在私有的buffer字段中，额外的重播缓存存储在replayCache字段中，两者的大小之和即为当前总缓冲大小。
• 针对flowOn产生的缓冲区：flowOn操作符生成的ChannelFlow内部持有私有的channel字段，反射读取该channel的bufferSize属性即可获得当前缓冲大小。需要注意冷流特性：每次Flow收集都会生成独立的新缓冲区，需针对每个收集实例单独反射获取。

方案2：埋点统计（生产环境推荐，稳定兼容所有版本） #

不需要依赖内部实现，通过自定义操作符统计上下游的元素收发差值即可得到缓冲区大小：

监控flowOn缓冲区大小示例： #

import java.util.concurrent.atomic.AtomicInteger

// 统计context1对应的flowOn缓冲区大小
val flowOn1BufferSize = AtomicInteger(0)

val businessFlow = flowOf(1, 2, 3)
    .map { /* 上游处理逻辑 */ }
    // 元素进入缓冲区前计数+1
    .onEach { flowOn1BufferSize.incrementAndGet() }
    .flowOn(context1)
    // 元素离开缓冲区后计数-1
    .onEach { flowOn1BufferSize.decrementAndGet() }
    .map { /* 下游处理逻辑 */ }
    .flowOn(context2)

监控MutableSharedFlow缓冲区大小示例： #

val sharedFlowBufferSize = AtomicInteger(0)
val sharedFlow = MutableSharedFlow<Int>(extraBufferCapacity = 32)

// 发射侧埋点
suspend fun emitData(data: Int) {
    sharedFlow.emit(data)
    sharedFlowBufferSize.incrementAndGet()
}

// 收集侧埋点
sharedFlow
    .onEach { sharedFlowBufferSize.decrementAndGet() }
    .collect { /* 业务处理逻辑 */ }

补充间接监控手段： #

如果不需要精确的缓冲区大小，仅需要判断缓冲压力，可配置buffer/flowOn/MutableSharedFlow的onBufferOverflow回调，统计溢出次数，间接评估缓冲区的负载情况。

官方设计说明 #

Flow被设计为声明式的数据流框架，内部缓冲实现属于可变更的底层细节（例如未来版本可能优化flowOn的实现，不再使用Channel作为缓冲载体），开放缓冲区访问API会限制框架的优化空间，因此官方没有计划提供相关公开接口。

内容的提问来源于stack exchange，提问作者Ondra K.