16、WorkManager与RxJava集成：RxWorker的createWork()方法、Single/Completable/Observable转换、RxJava中的背压与取消

说实话，WorkManager 本身已经很强大了。但如果你跟我一样，项目里早就用上了 RxJava，那肯定希望后台任务也能用响应式的方式去写。嗯，Google 也想到了这一点——他们提供了 RxWorker，让我们可以在 WorkManager 里直接使用 RxJava 的链式调用。

这一章，我就带你看看 RxWorker 到底怎么用。包括它的核心方法 createWork()，以及如何把 Single、Completable、Observable 转换成 WorkManager 能识别的结果。最后，我还会聊聊背压和取消这两个容易踩坑的点。

16.1 RxWorker 是什么？

RxWorker 是 WorkManager 提供的一个抽象类。它继承自 ListenableWorker，但把回调改成了 RxJava 的 Observable 形式。说白了，你不需要再手动处理 onStartJob() 和 onStopJob() 了，直接返回一个 Observable 就行。

我个人习惯把 RxWorker 理解为「WorkManager 和 RxJava 之间的桥梁」。你想想看，原本 WorkManager 的任务生命周期是回调驱动的，而 RxJava 是流驱动的。RxWorker 把这两者粘在了一起。

核心依赖： 使用 RxWorker 需要在 build.gradle 中添加： 
implementation "androidx.work:work-rxjava2:2.8.1"
注意版本号要和你的 WorkManager 主版本匹配。

16.2 createWork() 方法详解

RxWorker 只有一个抽象方法需要你实现：createWork()。它的签名是这样的：

public abstract Single<Result> createWork();

看到没？它返回的是一个 Single<Result>。Single 只发射一次数据，要么成功要么失败。这正好对应 WorkManager 的 Result：success、failure、retry。

我举个例子。假设你要在后台下载一个配置文件：

class ConfigDownloadWorker(
    context: Context,
    params: WorkerParameters
) : RxWorker(context, params) {

    override fun createWork(): Single<Result> {
        return downloadConfig()
            .map { Result.success() }
            .onErrorReturn { error ->
                Log.e("ConfigWorker", "下载失败", error)
                Result.retry()
            }
    }

    private fun downloadConfig(): Single<String> {
        // 模拟网络请求
        return Single.fromCallable {
            Thread.sleep(2000)
            "config_data"
        }
    }
}

这里有个细节：createWork() 是在后台线程执行的。WorkManager 默认的线程池会帮你调度。你不需要自己切线程。

我的经验： 我曾经在 createWork() 里直接做了耗时操作，忘了用 RxJava 的异步调度。结果任务虽然能跑，但会阻塞 Worker 的线程池。后来我养成了习惯：所有耗时操作都放到 Single.fromCallable 或者 Observable 的 subscribeOn 里。

16.3 Single / Completable / Observable 的转换

实际项目中，你拿到的可能不是 Single，而是 Completable 或者 Observable。这时候就需要做转换了。

16.3.1 Completable 转 Result

Completable 只关心完成与否，不发射数据。比如你只是删除一个本地文件：

override fun createWork(): Single<Result> {
    return deleteTempFile()
        .andThen(Single.just(Result.success()))
        .onErrorReturn { Result.failure() }
}

private fun deleteTempFile(): Completable {
    return Completable.fromAction {
        File(context.cacheDir, "temp.dat").delete()
    }
}

注意 andThen 这个操作符。它等 Completable 完成后，再执行后面的 Single。我个人觉得这是最优雅的写法。

16.3.2 Observable 转 Result

Observable 会发射多个数据。但 WorkManager 只需要一个结果。所以你要把多个数据「收拢」成一个 Result。

举个例子，你要批量上传图片：

override fun createWork(): Single<Result> {
    val imageUris = inputData.getStringArray("image_uris") ?: emptyArray()

    return Observable.fromArray(*imageUris)
        .flatMapSingle { uri -> uploadImage(uri) }
        .toList()
        .map { results ->
            if (results.all { it }) Result.success()
            else Result.failure()
        }
        .onErrorReturn { Result.retry() }
}

private fun uploadImage(uri: String): Single<Boolean> {
    // 模拟上传
    return Single.fromCallable { true }
}

这里用 toList() 把多个上传结果收集成一个 List，然后判断是否全部成功。嗯，你也可以用 reduce 或者 collect，看个人习惯。

注意： 如果 Observable 发射的数据量很大，比如几千张图片，直接用 toList() 可能会 OOM。这时候建议用 buffer 或者 window 分批处理。

16.4 RxJava 中的背压问题

背压这个词听起来挺吓人，说白了就是「上游发射太快，下游处理不过来」。在 WorkManager 的场景里，背压通常出现在你用 Observable 处理大量数据的时候。

举个例子，你要从数据库里读取一万条记录，然后逐条上传：

override fun createWork(): Single<Result> {
    return getRecordsFromDb()  // 返回 Flowable<Record>
        .onBackpressureBuffer()  // 加背压策略
        .flatMapSingle { record -> uploadRecord(record) }
        .toList()
        .map { Result.success() }
}

这里我用了 Flowable 而不是 Observable。为什么？因为 Flowable 支持背压，Observable 不支持。如果你用 Observable 处理大量数据，而且下游处理慢，就会抛出 MissingBackpressureException。

我曾经在一个日志上传任务里踩过这个坑。当时用 Observable 从文件里读日志，每行一条，结果日志文件有几十万行。程序跑了几分钟就崩了。后来换成 Flowable 加上 onBackpressureDrop 策略，才稳定下来。

常用的背压策略有：

策略 说明 适用场景
onBackpressureBuffer 把所有数据缓存起来 数据量不大，或者内存充足
onBackpressureDrop 丢弃处理不过来的数据 可以容忍丢数据，比如日志采样
onBackpressureLatest 只保留最新的数据 只关心最新状态，比如进度更新
我的建议： 在 WorkManager 里，如果数据量不确定，优先用 Flowable。Observable 只适合数据量小且确定不会背压的场景。你想想看，后台任务本来就跑在后台，出错了也不容易被发现，所以稳妥一点好。

16.5 任务的取消与生命周期

WorkManager 的任务是可以被取消的。比如用户退出了应用，或者你调用了 WorkManager.cancelWorkById()。这时候，RxWorker 需要能响应取消事件。

RxWorker 内部已经帮我们处理了大部分逻辑。当 WorkManager 要求停止任务时，RxWorker 会调用 dispose() 来取消订阅。但有个前提：你的 Observable 必须支持取消。

举个例子：

override fun createWork(): Single<Result> {
    return Single.create { emitter ->
        val thread = Thread {
            for (i in 0 until 100) {
                if (emitter.isDisposed) {
                    // 检查是否被取消
                    return@Thread
                }
                Thread.sleep(100)
                // 模拟工作
            }
            emitter.onSuccess(Result.success())
        }
        thread.start()

        // 设置取消回调
        emitter.setCancellable {
            thread.interrupt()
        }
    }
}

这里我手动创建了一个线程，并检查 emitter.isDisposed。如果任务被取消，就提前退出。同时用 setCancellable 注册了取消时的清理动作。

当然，如果你用的是 Retrofit 或者 Room 这类库，它们本身就支持取消。比如 Retrofit 的 Call 会在订阅取消时自动关闭。你不需要额外处理。

注意： 不要忘了处理 onStop() 方法。RxWorker 的 onStop() 会在任务被取消时调用。你可以在这里做一些资源清理，比如关闭数据库连接、释放文件句柄等。

16.6 知识体系总览

下面这张图总结了 RxWorker 的核心逻辑和关键节点：

RxWorker 核心逻辑流程图 WorkManager 调度 RxWorker.createWork() Single<Result> Completable → Single Observable → Single 背压策略（Flowable） 取消与生命周期 onStop() 清理 Result.success() / failure() / retry()

从这张图可以看出来，整个流程其实很清晰：WorkManager 调度 → RxWorker 执行 createWork() → 根据业务选择 Single/Completable/Observable 转换 → 处理背压和取消 → 返回 Result。每一步都有对应的处理策略。

16.7 小结

RxWorker 让 WorkManager 和 RxJava 的集成变得非常自然。你只需要关注 createWork() 方法，返回一个 Single<Result> 就行。其他事情，比如线程调度、生命周期管理，RxWorker 都帮你处理好了。

但有几个点你要记住：

  • Completable 用 andThen 转 Single
  • Observable 用 toList() 或 reduce 收拢数据
  • 大量数据时用 Flowable 加背压策略
  • 任务取消时检查 isDisposed，并在 onStop() 里清理资源

嗯，这些经验都是我一个个坑踩出来的。希望你能少走弯路。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →