RxJava 迁移至 Kotlin Flow 的背压策略对比与实现

RxJava 迁移至 Kotlin Flow 的背压策略对比与实现 | 极客日志

RxJava 迁移至 Kotlin Flow 的背压策略对比与实现

前言

背压（Backpressure）是响应式编程中一个非常复杂且关键的话题。本文旨在分析从 RxJava 迁移到 Kotlin Flow 过程中，如何利用各自的背压方案解决数据流控制问题。由于技术实现的差异及协程内部的复杂性，本文将通过示例对比两者在背压处理上的不同策略，帮助开发者理解迁移过程中的核心变化。

注意：本文侧重于从 RxJava 的角度出发，对比 Flow 背压的差异和相关策略的使用方案。

关于 RxJava 的背压机制

在 RxJava 中，最核心的流类型是 Observable。默认情况下，Observable 没有提供优雅的方式来处理背压，所有发送/接收的数据都存储在内存中并确保订阅者接收到它。如果发送的数据量非常大，最终可能导致 OutOfMemoryError (OOM) 甚至程序崩溃。

在 RxJava 中，只有 Flowable 流类型才原生支持背压处理。它默认具有 128 大小的缓冲区，并通过 Subscriber 接口支持背压管理。

Subscriber 接口结构

public interface Subscriber<T> { 
    public void onSubscribe(Subscription s); 
    public void onNext(T t); 
    public void onError(Throwable t); 
    public void onComplete(); 
}

其中，Subscription 接口负责提供背压管理功能：

public interface Subscription {
    public void request(long n);
    public void cancel();
}

request 方法至关重要：每当下游能够处理更多事件时，它会向上游发送请求，告知上游它能消费的事件数量。

基本通信流程

下游订阅时请求一些事件（例如 1 个）。
上游收到请求并产生下一个事件。
下游接收到事件后，可以再次向上游请求更多事件。

这种机制意味着：如果下游没有请求事件，生产者应该停止生产。如果应用了某种背压策略且生产者能发出新项目而消费者无法消费，生产者可能会丢弃值 (Drop) 或缓冲它们 (Buffer)。这被视为一种从链底到顶部的通信，告诉上游是否发出值，并对未准备好发送给消费者的值应用特定规则。

关于 Flow 的背压机制

在 Kotlin Flow 中，背压处理机制有所不同。Flow 没有像 RxJava 那样从下游到上游的直接链式通信（Request-Driven）。一切基于 suspend 挂起函数。

当流的收集器（Collector）不堪重负时，它可以简单地暂停发射器（Emitter），然后在准备好接受更多元素时恢复它。当下游暂停或执行其他工作时，上游会识别到并发射器不会继续发射元素。这种机制更依赖于协程的协作式调度。

RxJava 与 Flow 的简单比较

为了直观比较两者的背压策略，我们通过测试用例模拟从网络获取数据等场景。简单来说，RxJava 倾向于阻塞线程等待处理，而 Flow 则倾向于挂起协程。

基础流定义对比

对于 RxJava，我们可以这样定义延迟发射的流：

private fun flowable(delay: Long, mode: BackpressureStrategy, limit: Int): Flowable<Int> =
        Flowable.create<Int>({ emitter ->
            for (i in 1..limit) {
                Thread.sleep(delay)
                emitter.onNext(i)
            }
        }, mode)

对于 Flow，等效的策略代码如下：

private fun flow(timeout: Long, limit: Int): Flow<Int> = flow {
        for (i in 1..limit) {
            delay(timeout)
            emit(i)
        }
    }

上述代码实现了在每个事件之间延迟 timeout 时长，并发出 limit + 1 个项目的流。

验证测试

为了详细验证，我们补充代码让上游每 100 毫秒生产一次项目，并依次用 200 毫秒和 300 毫秒处理它们。通过这样的设置，预计并非所有事件都会被消费，从而触发背压。

RxJava 测试代码

fun testFlowable(mode: BackpressureStrategy, limit: Int = 10) {
    val latch = CountDownLatch(1)
    val stringBuffer = StringBuffer()
    val time = System.currentTimeMillis()

    flowable(delay = 100, mode = mode, limit = limit)
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.computation(), false, 1)
        .map { doWorkBlocking(i = it, delay = 200) }
        .map { doWorkBlocking(i = it, delay = 300) }
        .doOnComplete {
            latch.countDown()
        }
        .subscribe {
            stringBuffer.append("$it")
        }

    latch.await()
    println(System.currentTimeMillis() - time)
    println(stringBuffer.toString())
}

Flow 测试代码

fun testFlow(limit: Int = 10, onBackpressure: Flow<Int>.() -> Flow<Int>) {
    val latch = CountDownLatch(1)
    val time = System.currentTimeMillis()
    val stringBuffer = StringBuffer()

    CoroutineScope(Job() + Dispatchers.Default).launch {
        flow(timeout = 100, limit = limit)
            .flowOn(Dispatchers.IO)
            .onBackpressure()
            .map { doWorkDelay(i = it, timeout = 200) }
            .map { doWorkDelay(i = it, timeout = 300) }
            .onCompletion { latch.countDown() }
            .collect {
                stringBuffer.append("$it")
            }
    }

    latch.await()
    println((System.currentTimeMillis() - time))
    println(stringBuffer.toString())
}

RxJava 的背压策略详解

RxJava Flowable 提供了多种背压策略，以下是主要策略的说明与实践。

1. Buffer (缓冲)

策略说明： 缓冲生产者的所有项目。注意，如果数据量过大，可能导致 OOM 内存泄漏。

使用场景： 当数据生成速度快于消耗速度时，为观察者提供一些时间来赶上生产过剩的来源。可以通过调用 buffer() 方法来定义缓冲区大小。

val source = PublishSubject.create<Int>()
source.buffer(1024)
    .observeOn(Schedulers.computation())
    .subscribe(ComputeFunction::compute, Throwable::printStackTrace)

定义大小为 1024 的缓冲区会给观察者一些时间。如果生产者过度生成元素超出了预测的缓冲区大小，溢出仍然可能发生，因此这通常只是一个临时修复方案。

`.onBackpressureBuffer()` 实践

我们希望接收发出的事件，即使消费者跟不上。RxJava 可以缓冲项目直到消费者准备好处理它们（可能导致 OutOfMemoryException），而 Flow 可以暂停发射器。

RxJava 示例：

fun rxBuffer() {
    Flowable.range(1, 10)
        .onBackpressureBuffer()
        .observeOn(Schedulers.single())
        .subscribe { value ->
            Thread.sleep(100)
            println("Get value: $value")
        }
    Thread.sleep(1000)
}

Flow 等效策略：

fun main() = runBlocking {
    (1..1_000_000).asFlow()
        .buffer(capacity = 0, onBufferOverflow = BufferOverflow.SUSPEND)
        .collect { value ->
            delay(100)
            println("Got value: $value")
        }
}

在 Flow 中，默认情况下会暂停发射器。当发射器和收集器在单独的协程中运行时，如果它们都需要一些时间才能完成，它们可以并发执行，这样速度会更快。

2. Drop (丢弃)

策略说明： 如果消费者无法跟上并且缓冲区已满，此策略将丢弃所有发出的项目。类似于 Latest，但不会保留最新数据，而是直接丢弃。

RxJava 示例：

fun rxDrop() {
    Flowable.range(1, 10)
        .onBackpressureDrop()
        // 将默认的缓冲区 size 改为 1
        .observeOn(Schedulers.single(), false)
        .subscribe { value ->
            Thread.sleep(100)
            println("Get Value: $value")
        }
    Thread.sleep(1000)
}

可以看到输出结果只拿到了一个 1，之后的所有值都被丢弃了，因为消费者无法处理它们。

注意事项：

如果没有更改 Scheduler 调度器，代码将同步运行，消费者会阻塞生产者，此时可能获得所有项目。
observeOn() 中使用默认缓冲区大小是 128，指定为 1 时才会触发丢弃逻辑。

Flow 等效策略：

fun flowDrop() = runBlocking {
    (1..10).asFlow()
        .buffer(capacity = 1, onBufferOverflow = BufferOverflow.DROP_LATEST)
        .collect { value ->
            delay(100)
            println("Get value: $value")
        }
}

与 RxJava 不一样的是，鉴于协程的性质，收集器处理前两个元素，但想法相同——所有未处理的元素都被丢弃。为了使缓冲和背压处理正常工作，我们需要在单独的协程中运行收集器，这就是 .buffer() 操作符的作用，它将所有发出的项目发送到在单独协程中运行的收集器。

3. Latest (最新)

策略说明： 在背压的情况下只发出最新的项目。丢弃掉之前的数据，只保留生产者给的最新的数据。

RxJava 示例：

fun rxLatest() {
    Flowable.range(1, 10)
        .onBackpressureLatest()
        .observeOn(Schedulers.single(), false, 2)
        .subscribe { value ->
            Thread.sleep(100)
            println("Get value:$value ")
        }
    Thread.sleep(1000L)
}

我们将缓冲区大小增加到 2。在这种情况下，缓冲区将填充最旧的值，并删除所有后续值，最后一个值除外。

Flow 等效策略：

fun flowLatest() = runBlocking {
    (1..10).asFlow()
        .buffer(capacity = 2, onBufferOverflow = BufferOverflow.DROP_OLDEST)
        .collect { value ->
            delay(100)
            println("Get Value: $value")
        }
}

在 Flow 中，新数据会直接覆盖掉旧数据，不设缓冲区（即缓冲区大小为 0），丢弃旧数据。Flow 中对此也提供了相应的操作符 conflate。

迁移建议与注意事项

从 RxJava 迁移到 Kotlin Flow 不仅仅是语法的改变，更是思维模式的转变。以下是迁移过程中需要注意的关键点：

1. 阻塞 vs 挂起

RxJava 大量依赖线程阻塞（如 Thread.sleep），而 Flow 依赖挂起函数（如 delay）。在迁移时，务必将阻塞操作替换为非阻塞的协程挂起操作，以避免占用不必要的线程资源。

2. 调度器的映射

RxJava 中的 Schedulers.io()、Schedulers.computation() 需要映射到 Flow 的 Dispatchers.IO、Dispatchers.Default。同时，Flow 的 flowOn 操作符用于指定协程上下文，这与 RxJava 的 subscribeOn 类似但行为略有不同。

3. 背压策略的选择

Buffer: 适用于数据量可控且必须保证不丢失数据的场景。需注意内存限制。
Drop: 适用于实时性要求高，允许丢失部分历史数据的场景（如传感器数据流）。
Latest: 适用于只需要最新状态的场景（如 UI 状态更新）。

4. 错误处理

RxJava 使用 onError 回调，Flow 使用 catch 操作符或 try-catch 块。迁移时需确保错误传播路径一致。

5. 生命周期管理

RxJava 依赖 Disposable 管理订阅，Flow 依赖 CoroutineScope 和 Job。确保在组件销毁时正确取消协程，防止内存泄漏。

总结

RxJava 库提供了丰富的背压处理功能，灵活性高，但学习曲线陡峭，正确使用这些操作对大多数开发人员来说并不是一项微不足道的任务。相比之下，Kotlin Flow 基于协程的 suspend 机制，背压处理更加直观且易于理解。

总的来说，RxJava Flowable 和 Kotlin Flow 都支持背压，但机制存在差异。RxJava 内置的背压支持是从下到上工作的（下游能够在需要更多值时告诉上游），而 Kotlin Flow 背压主要基于 suspend 挂起函数。当我们从 RxJava 迁移到 Kotlin Flow 时，请特别注意使用 Flowable 和使用到背压策略的地方。Kotlin Flow 对初学者更友好，同时为您可能遇到的大多数问题提供了解决方案。在实际项目中，根据业务需求选择合适的背压策略，并充分利用协程的特性进行优化，是提升应用性能的关键。

RxJava 迁移至 Kotlin Flow 的背压策略对比与实现