当前位置：首页 > news >正文

典型的高频、实时、多消费者的事件流场景。 --＞SharedFlow

news 2025/10/8 14:34:07

对于识别人脸夹角、人脸跟随这种实时计算机视觉场景，我强烈推荐使用 SharedFlow。这是一个典型的高频、实时、多消费者的事件流场景。

为什么 SharedFlow 是最佳选择

1. 场景特点分析

高频数据: 人脸检测通常 15-60 FPS
多消费者: 可能需要同时显示、记录、分析
实时性要求高: 延迟要尽可能低
新数据更重要: 过时的帧可以丢弃

2. 推荐方案：SharedFlow

class FaceTrackingManager {// 使用 SharedFlow 处理人脸数据流private val _faceDataFlow = MutableSharedFlow<FaceData>(replay = 0,                    // 不重放旧数据extraBufferCapacity = 2,       // 小缓冲区，降低延迟onBufferOverflow = BufferOverflow.DROP_OLDEST // 丢弃旧帧，保证实时性)val faceDataFlow: SharedFlow<FaceData> = _faceDataFlow.asSharedFlow()// 处理摄像头帧并发布人脸数据fun processFrame(frame: Frame) {val faceData = detectFace(frame)// 使用 tryEmit 避免挂起，保证实时性_faceDataFlow.tryEmit(faceData)}private fun detectFace(frame: Frame): FaceData {// 人脸检测逻辑val angle = calculateFaceAngle(frame)val position = detectFacePosition(frame)return FaceData(angle, position, System.currentTimeMillis())}
}

3. 完整的架构示例

// 数据模型
data class FaceData(val angle: Float,      // 人脸夹角val position: Rect,    // 人脸位置val timestamp: Long,   // 时间戳val frameId: Int = 0   // 帧ID
)class FaceTrackingSystem {private val trackingManager = FaceTrackingManager()private var frameCounter = 0// 启动人脸跟踪fun startTracking(cameraSource: CameraSource) {// 订阅人脸数据setupSubscribers()// 处理摄像头帧cameraSource.setFrameListener { frame ->trackingManager.processFrame(frame.copy(id = frameCounter++))}}private fun setupSubscribers() {// 消费者1: 显示实时角度CoroutineScope(Dispatchers.Main).launch {trackingManager.faceDataFlow.collect { faceData ->updateAngleDisplay(faceData.angle)}}// 消费者2: 控制云台跟随CoroutineScope(Dispatchers.IO).launch {trackingManager.faceDataFlow.collect { faceData ->controlGimbal(faceData.position)}}// 消费者3: 记录分析数据CoroutineScope(Dispatchers.Default).launch {trackingManager.faceDataFlow.sample(100) // 每100ms采样一次，避免过于频繁.collect { faceData ->logFaceData(faceData)}}// 消费者4: 触发警报（当角度异常时）CoroutineScope(Dispatchers.IO).launch {trackingManager.faceDataFlow.filter { it.angle > 30.0f } // 只处理角度过大的情况.collect { faceData ->triggerAngleAlert(faceData.angle)}}}private fun updateAngleDisplay(angle: Float) {// 更新UI显示angleTextView.text = "角度: ${"%.1f".format(angle)}°"}private fun controlGimbal(position: Rect) {// 控制云台跟随人脸gimbalController.moveTo(position.centerX(), position.centerY())}private fun logFaceData(faceData: FaceData) {// 记录到数据库或文件database.insertFaceData(faceData)}private fun triggerAngleAlert(angle: Float) {// 触发角度异常警报alertSystem.notify("人脸角度异常: ${"%.1f".format(angle)}°")}
}

4. 性能优化版本

对于高性能要求的场景：

class HighPerformanceFaceTracker {private val _faceDataFlow = MutableSharedFlow<FaceData>(replay = 0,extraBufferCapacity = 1,  // 最小缓冲区，最低延迟onBufferOverflow = BufferOverflow.DROP_OLDEST)// 使用专门的调度器private val visionScope = CoroutineScope(Dispatchers.Default + SupervisorJob())private val uiScope = CoroutineScope(Dispatchers.Main + SupervisorJob())// 批量处理以提高性能private val batchFlow = _faceDataFlow.buffer(16) // 适当的缓冲区.map { data -> // 数据预处理preprocessFaceData(data) }.shareIn(scope = visionScope,started = SharingStarted.WhileSubscribed(5000),replay = 0)fun setupProcessing() {// 高性能消费者 - 在专用线程池中运行visionScope.launch {batchFlow.collect { faceData ->performHeavyAnalysis(faceData)}}// UI更新消费者 - 在主线程运行uiScope.launch {batchFlow.sample(16) // 限制UI更新频率.collect { faceData ->updateUI(faceData)}}}
}

5. 与 Channel 的对比

特性	SharedFlow (推荐)	Channel (不推荐)
多消费者	✅ 所有消费者收到所有数据	❌ 一个消息只被一个消费者处理
内存使用	✅ 可配置缓冲区，可丢弃旧数据	⚠️ 可能堆积消息
实时性	✅ 可配置为低延迟模式	⚠️ 消费者阻塞会影响实时性
背压处理	✅ 灵活的背压策略	⚠️ 发送可能挂起
资源管理	✅ 自动清理无订阅者	⚠️ 需要手动管理

6. 配置建议

根据你的具体需求调整 SharedFlow 配置：

// 低延迟模式 (人脸跟随)
val lowLatencyFlow = MutableSharedFlow<FaceData>(replay = 0,extraBufferCapacity = 1,onBufferOverflow = BufferOverflow.DROP_OLDEST
)// 平衡模式 (一般使用)
val balancedFlow = MutableSharedFlow<FaceData>(replay = 0,extraBufferCapacity = 8,onBufferOverflow = BufferOverflow.DROP_OLDEST
)// 保证数据模式 (数据分析)
val guaranteedFlow = MutableSharedFlow<FaceData>(replay = 0,extraBufferCapacity = 32,onBufferOverflow = BufferOverflow.SUSPEND
)