使用 RxJava flowable 处理 Android Room 查询时提高性能

Question

我的应用正在以可能的最高采样率（在我的设备上约为 200 Hz）从加速度计收集传感器值，并将这些值保存在 Room 数据库中。我还想经常用最新的测量值更新一些图表，比如说每秒 5 次的刷新率。自从该应用程序还收集了大约 200 Hz 的线性加速度（没有 g）（因此两个传感器每个都以大约 200Hz 的频率将值插入数据库），我注意到应用程序性能大幅下降，我有几秒钟的延迟在收集的加速度值和它们显示在图中的值之间。 根据分析器，我的猜测是 RxComputationThread 是瓶颈，因为由于 Flowables，它几乎一直处于活动状态。

我使用 sample() 来限制接收器更新，因为我的图表不需要非常频繁地更新。当我只收集一个传感器时，这导致了可接受的性能。我看到 RxJava 提供了一个 interval() 方法来限制发射器端的发射频率，但这对我来说似乎不可用？ （未解决的参考）。

也许有人知道如何提高性能？总的来说，我喜欢 RxJava 和 Room 的概念，并希望坚持使用它们，但我几乎被困在这一点上。

这是我用来观察 Room SQL 表并更新图表的代码：

// Observe changes to the datasource and create a new subscription if necessary
sharedViewModel.dataSource.observe(viewLifecycleOwner, Observer { source ->
    Log.d("TAG", "Change observed!")
    when (source) {
        "acc" -> {
            val disposableDataSource =
                sharedViewModel.lastSecondsAccelerations
                    .sample(200, TimeUnit.MILLISECONDS)
                    .onBackpressureDrop()
                    .subscribeOn(Schedulers.io())
                    .subscribe { lastMeasurements ->
                        Log.d("TAG", Thread.currentThread().name)
                        if (sharedViewModel.isReset.value == true && lastMeasurements.isNotEmpty()) {
                            val t =
                                lastMeasurements.map { (it.time.toDouble() * 1e-9) - (lastMeasurements.last().time.toDouble() * 1e-9) }
                            val accX = lastMeasurements.map { it.accX.toDouble() }
                            val accY = lastMeasurements.map { it.accY.toDouble() }
                            val accZ = lastMeasurements.map { it.accZ.toDouble() }

                            // Update plots
                            updatePlots(t, accX, accY, accZ)
                        }
                    }
            compositeDisposable.clear()
            compositeDisposable.add(disposableDataSource)
        }
        "lin_acc" -> {
            val disposableDataSource =
                sharedViewModel.lastSecondsLinAccelerations
                    .sample(200, TimeUnit.MILLISECONDS)
                    .onBackpressureDrop()
                    .subscribeOn(Schedulers.io())
                    .subscribe { lastMeasurements ->
                        Log.d("TAG", Thread.currentThread().name)
                        if (sharedViewModel.isReset.value == true && lastMeasurements.isNotEmpty()) {
                            val t =
                                lastMeasurements.map { (it.time.toDouble() * 1e-9) - (lastMeasurements.last().time.toDouble() * 1e-9) }
                            val accX = lastMeasurements.map { it.accX.toDouble() }
                            val accY = lastMeasurements.map { it.accY.toDouble() }
                            val accZ = lastMeasurements.map { it.accZ.toDouble() }

                            // Update plots
                            updatePlots(t, accX, accY, accZ)
                        }
                    }
            compositeDisposable.clear()
            compositeDisposable.add(disposableDataSource)
        }
    }
})

获取最后 10 秒测量值的查询

@Query("SELECT * FROM acc_measurements_table WHERE time > ((SELECT MAX(time) from acc_measurements_table)- 1e10)")
fun getLastAccelerations(): Flowable<List<AccMeasurement>>

Answer 1

感谢您的 cmets，我现在想通了，瓶颈是什么。问题是大量的插入调用，这并不奇怪。但是可以通过使用某种缓冲区一次插入多行来提高性能。

这是我添加的，以防有人在相同的情况下运行：

class InsertHelper(private val repository: Repository){
    var compositeDisposable = CompositeDisposable()

    private val measurementListAcc: FlowableList<AccMeasurement> = FlowableList()
    private val measurementListLinAcc: FlowableList<LinAccMeasurement> = FlowableList()

    fun insertAcc(measurement: AccMeasurement) {
        measurementListAcc.add(measurement)
    }
    fun insertLinAcc(measurement: LinAccMeasurement) {
        measurementListLinAcc.add(measurement)
    }

    init {
        val disposableAcc = measurementListAcc.subject
            .buffer(50)
            .subscribe {measurements ->
                GlobalScope.launch {
                    repository.insertAcc(measurements)
                }
                measurementListAcc.remove(measurements as ArrayList<AccMeasurement>)
            }

        val disposableLinAcc = measurementListLinAcc.subject
            .buffer(50)
            .subscribe {measurements ->
                GlobalScope.launch {
                    repository.insertLinAcc(measurements)
                }
                measurementListLinAcc.remove(measurements as ArrayList<LinAccMeasurement>)
            }

        compositeDisposable.add(disposableAcc)
        compositeDisposable.add(disposableLinAcc)
    }
}

// Dynamic list that can be subscribed on
class FlowableList<T> {
    private val list: MutableList<T> = ArrayList()
    val subject = PublishSubject.create<T>()

    fun add(value: T) {
        list.add(value)
        subject.onNext(value)
    }

    fun remove(value: ArrayList<T>) {
        list.removeAll(value)
    }
}

我基本上使用一个动态列表来缓冲几十个测量样本，然后将它们作为一个整体插入到房间数据库中，然后从动态列表中删除它们。这里还有一些为什么批量插入更快的信息：https://hackernoon.com/squeezing-performance-from-sqlite-insertions-with-room-d769512f8330

我对 Android 开发还是很陌生，所以如果您发现一些错误或有建议，我感谢您的每一条评论 :)