📚 课程目录第4章数据流架构

4. 数据流架构：生产者-消费者模式、背压处理、数据持久化策略

埋点数据在模块间流转时，核心矛盾在于生产速度与消费速度的不匹配。本章以生产者-消费者模式为骨架，定义模块边界与分层职责，并通过背压处理和数据持久化保证系统的稳定性与可靠性。

4.1 模块边界与分层职责

整个数据流划分为三层，每层职责明确、边界清晰：

层级	角色	职责
生产层	生产者（模块A、B、C …）	采集埋点事件，封装为统一格式的数据包，推入共享通道。
流转层	缓冲区 + 控制器	接收事件、暂存、背压感知、触发持久化或丢弃策略。
消费层	消费者（聚合、过滤、上报、存储）	从缓冲区取出事件，执行业务逻辑（去重、标记、序列化）。

  关键边界：生产者只负责“推数据”，不关心消费状态；消费者只处理“从通道取到的数据”，不反向干扰生产。信任链路建立在通道协议上——生产者必须遵循Event数据结构的约定，消费者不信任任何模块的原始状态。

4.2 生产者-消费者设计

在 Android 环境中，我们使用 Kotlin Flow 作为底层通道，通过广播式（BroadcastChannel）或共享流（SharedFlow）实现一对多生产。生产者侧封装为统一的EventProducer接口：

interface EventProducer {
    // 所有生产者通过此接口提交事件
    suspend fun produce(event: Event)
}

// 典型实现：委托给一个 MutableSharedFlow
class SharedFlowEventProducer : EventProducer {
    private val _eventFlow = MutableSharedFlow<Event>(
        replay = 0,
        extraBufferCapacity = 64,
        onBufferOverflow = BufferOverflow.SUSPEND
    )

    val eventFlow: Flow<Event> get() = _eventFlow.asSharedFlow()

    override suspend fun produce(event: Event) {
        _eventFlow.emit(event)
    }
}

消费者侧使用 Flow.collect 并配合背压策略。设计时需注意：不要在主线程生产或消费；消费者可并行运行但需加锁保护共享资源。

4.3 背压处理

当消费者处理速度低于生产者推入速度时，背压必须显式处理，否则导致 OOM 或数据丢失。我们采用 两层背压 策略：

缓冲区背压：设置 extraBufferCapacity 和 onBufferOverflow 参数。当缓冲区满时，SUSPEND 模式会挂起生产者协程，实现自然反压。
持久化降级：若缓冲区持续满载超过阈值（例如连续 5 次 suspend 超过 100ms），触发持久化降级——新事件直接写入 SQLite 数据库，消费者继续从内存和数据库双源消费。

实践建议：不要使用 DROP_OLDEST 或 DROP_LATEST 作为默认背压策略，因为它们会导致数据不可控流失。仅在低优先级埋点（如非关键日志）且明确文档化时才可降级。

背压反馈机制在通道层隐式实现：生产者通过 emit 的挂起感知背压，无需显式回调。如果需要统计背压次数，可在 SharedFlow 外部包一层 Flow 转换：

fun <T> Flow<T>.withBackPressureMonitor(): Flow<T> {
    var suspendedCount = 0
    // 实际生产中使用 snapshotFlow 或 callbackFlow
    return this.onStart {
        // 初始化监测
    }
}

4.4 数据持久化策略

持久化不是为了长期存储，而是作为背压降级和进程重启保障。需在实时性和性能之间权衡：

策略	存储介质	写入时机	读取方式	风险
全量持久化	SQLite	每产生一个事件即写入	消费者优先读内存，其次读 DB	磁盘 IO 瓶颈，App 卡顿；需异步 WAL 模式
批量持久化	文件（JSON/序列化）	背压触发或定时（5s/50事件）	启动时反序列化到内存缓冲区	进程崩溃丢失最后一批事件
双缓冲	内存 + SQLite	背压触发写入，正常时仅写内存	消费者从内存读取，DB 作为冗余	实现复杂度增加；需处理 DB 与内存的数据去重

风险点：不要在主线程执行任何持久化操作。即使使用 Room 的 suspend 方法，也应在 Dispatchers.IO 上下文中调用。另外，持久化后的数据如果在重启时消费完毕，必须清理旧文件/表，否则累积导致存储膨胀。

4.5 选型权衡

SharedFlow vs. Channel
SharedFlow 支持多个独立订阅、回放功能（可配置 replay），适合一对多；Channel 适合一对一且需要背压显式控制。推荐 SharedFlow + 自定义背压降级。
SQLite vs. 文件
SQLite 支持事务和查询，适合需要去重、排序的场景；文件写入快，但读取后清理复杂。首选 SQLite + WAL 模式。
SUSPEND vs. DROP
数据完整性要求高时用 SUSPEND，可接受丢数据时用 DROP_LATEST。埋点方案必须明确数据可靠性等级。

4.6 落地步骤

定义 Event 数据协议（模块边界契约）。
实现 SharedFlowEventProducer 并注入到各模块。
构建消费者 collect 逻辑，配置 extraBufferCapacity=64、onBufferOverflow=SUSPEND。
实现背压监测（记录 emit 挂起次数），达到阈值时开启降级持久化。
集成 Room/WAL 数据库，封装 EventDao 用于写入和批量读取。
编写单元测试验证：消费慢时生产者挂起、降级后数据不丢失。

4.7 验收标准

✅ 生产者每秒 1000 事件持续 10 秒，消费者处理速度 200/秒，无 OOM、无数据丢失（检查最终上报数量）。
✅ 模拟进程被 kill，重启后回放持久化数据，事件不漏。
✅ 背压降级后生产者协程不抛出 CancellationException 之外的异常。
✅ 主线程无磁盘读写、无 emit 调用。