使用RxJava 2处理Android微秒级传感器实时数据的计数、比较与存储问题

现有代码的核心问题排查 #

对着现有实现逐行捋，所有异常现象都是几个硬bug叠出来的：

• 浮点数直接用==做等值判断，传感器的a值受精度波动影响，几乎不可能刚好卡到3.9这个精确值，分支根本进不去——这就是日志打不出来、计数逻辑不触发的最大原因
• alphaValuesArray写在if分支内部，每进入一次判断就新建一个只装了当前值的数组，之前存的历史数据直接被冲没，根本攒不下数据完成累计存储、求解最大值的需求
• 计数逻辑没有做线程同步，直接在传感器回调线程跨线程操作Fragment UI、触发toneGenerator，多线程读写抢资源就会出现计数结果跳变、tone被异常触发的问题
• 时间差计算前没有做空值校验，Tbuffer_firstReading要是还没初始化、或者被意外重置，算出来的时间差完全没有业务意义

核心功能的实现优化 #

1. 实时数据存储与最大值计算 #

• 把存储数据的可变容器、当前最大值变量从方法内部提升为类成员变量，避免每次进入逻辑就重置数据
• 替换浮点数等值判断为带允许误差的范围匹配，兼容传感器的精度波动
• 维护固定时间窗口的边界逻辑，每到时间窗口阈值就重置存储列表和最大值，保证数据范围正确
  示例代码：

// 类级成员，不要放在业务方法内部
private val alphaValueWindow = mutableListOf<Double>()
private var currentWindowMax = Double.MIN_VALUE
private var windowStartTimestamp: Long? = null

private fun handleAlphaEvent(eventData: EventData) {
    // 浮点数精度兼容判断，替代直接的==比较
    if (abs(eventData.a - 3.9) >= 0.001) return

    // 窗口第一次进入时初始化时间起点
    if (windowStartTimestamp == null) {
        windowStartTimestamp = eventData.received_at
        alphaValueWindow.clear()
        currentWindowMax = Double.MIN_VALUE
    }

    // 满足200ms时间差要求就存入窗口
    val timeDiff = eventData.received_at - (windowStartTimestamp ?: return)
    if (timeDiff >= 200000) {
        alphaValueWindow.add(eventData.a)
        // 实时更新最大值，不需要每次遍历全列表计算
        if (eventData.a > currentWindowMax) {
            currentWindowMax = eventData.a
        }
        Log.d("SensorProcess", "存入窗口值: ${eventData.a}, 当前最大值: $currentWindowMax")
    }
}

2. 计数逻辑与Fragment展示、tone触发优化 #

• 用原子变量维护计数结果，避免多线程读写带来的计数偏差
• 把所有UI更新、tone触发的逻辑通过RxJava调度器切到主线程执行，不要在传感器回调线程直接操作UI
• 对计数结果做distinctUntilChanged过滤，只有计数真实变化的时候才发事件，避免重复刷新UI、重复触发tone
  示例操作符链实现：

// 原子计数变量，类级别初始化
private val matchCount = AtomicInteger(0)

sensorDataFlowable
    .onBackpressureBuffer() // 保留原有背压配置
    .observeOn(Schedulers.io())
    .filter { it != null }
    .map { eventData ->
        // 在IO线程完成所有数据计算、存储逻辑
        handleAlphaEvent(eventData)
        if (meetTriggerCondition(eventData)) {
            matchCount.incrementAndGet()
        }
        matchCount.get()
    }
    .distinctUntilChanged() // 过滤重复的相同计数结果
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { currentCount ->
        // 主线程更新Fragment的计数展示
        boundFragment?.updateCountText(currentCount)
        // 只有计数达到固定阈值时才触发tone，不要和数据存储逻辑混在一起
        if (currentCount == TONE_START_THRESHOLD) {
            toneGenerator.play()
        }
    }

修复后的效果说明 #

• 解决浮点数匹配失败的问题后，处理逻辑能正常进入分支，对应的日志可以稳定输出
• 数组存储从方法内的临时变量改成类级别的可变列表后，可以正常累计窗口内的实时数据，最大值计算的结果也会准确
• 计数逻辑加了原子更新、线程切换、结果去重后，Fragment展示的计数不会出现错乱，toneGenerator也只会在满足条件时被触发一次

注意：微秒级粒度的数据流不需要每次都把完整数组传到UI层，只需要把当前最大值、当前计数这两个需要展示的值切到主线程就行，避免大量数据跨线程传输带来的性能损耗

内容的提问来源于stack exchange，提问作者Satyam Raikwar