RxKotlin 官方指南：RxJava 的 Kotlin 扩展函数库

• 异步编程
• 后端

导读

RxKotlin 是一个轻量级库，为 RxJava 提供便捷的 Kotlin 扩展函数。在 Kotlin 中直接使用 RxJava 虽可行，但 Kotlin 的扩展函数等语言特性可以进一步简化 RxJava 的使用体验。本文将以 RxKotlin 官方 README 为骨架，结合仓库源码与测试用例，全面讲解 RxKotlin 的核心价值、快速上手、依赖引入、扩展函数目录、SAM 适配器、与 RxAndroid 等库的组合使用以及贡献指南。读完本文，你将掌握 RxKotlin 的完整功能清单、典型调用方式，并能将其无缝集成到自己的 Kotlin 项目中。

一、RxKotlin 是什么

RxKotlin 是一个轻量级库，专门为 RxJava 添加便捷的 Kotlin 扩展函数。你开箱即可用 Kotlin 使用 RxJava，但 Kotlin 的扩展函数等语言特性可以更流畅地简化 RxJava 的使用。RxKotlin 旨在保守地将这些便捷特性集中到一个库中，并规范在 Kotlin 中使用 RxJava 的惯例。

从源码结构看，RxKotlin 的核心实现全部位于 src/main/kotlin/io/reactivex/rxjava3/kotlin 目录下，是一组相互独立的 Kotlin 文件与顶层扩展函数，不引入任何复杂的运行时框架。

入门示例：以下代码是 README 中最经典的示例，演示了扩展函数和具名参数的结合：

import io.reactivex.rxjava3.kotlin.subscribeBy import io.reactivex.rxjava3.kotlin.toObservable fun main() { val list = listOf("Alpha", "Beta", "Gamma", "Delta", "Epsilon") list.toObservable() // extension function for Iterables .filter { it.length >= 5 } .subscribeBy( // named arguments for lambda Subscribers onNext = { println(it) }, onError = { it.printStackTrace() }, onComplete = { println("Done!") } ) }

该示例对应的完整测试位于 src/test/kotlin/io/reactivex/rxjava3/kotlin/ExtensionTests.kt 中的testFilter、testMap2等用例，它们验证了list.toObservable()与后续filter { }、map { }操作符的端到端行为。

内置风格约束：由于 Kotlin 使得为任何东西实现扩展都变得很容易，本项目必须保守地界定哪些特性在范围之内。意图创造语法糖的尝试会很快退化成"语法糖精"（过度语法糖），这种个人偏好更适合留在自己的私有领域，而非一个开源库中。

二、快速上手：添加依赖

RxKotlin 在 Maven Central 上发布，Maven、Ivy、Gradle 等构建工具都可以从中获取二进制包与依赖信息（可访问 http://search.maven.org 查询）。

2.1 版本与 RxJava 版本对应关系

重要提示：RxKotlin 3.x 对应 RxJava 3.x（当前活跃版本）；RxKotlin 2.x 对应 RxJava 2.x（维护模式，仅接受 bugfix，不再新增特性或行为变更）；RxKotlin 1.x 对应 RxJava 1.x（已到达生命终点，不再进行开发、支持、维护、PR 或更新）。

特别注意：维护者并不会随 RxJava 的每一个 minor 或 patch 版本同步更新 RxKotlin 的依赖版本，因此你应当显式地在自己的pom.xml或build.gradle(.kts)中添加所需的 RxJava 依赖版本。从仓库的 build.gradle.kts 可以看到当前 RxKotlin 3.x 分支实际声明了对io.reactivex.rxjava3:rxjava:3.0.6的api依赖。

2.2 RxKotlin 3.x（对应 RxJava 3.x）

Maven：

<dependency> <groupId>io.reactivex.rxjava3</groupId> <artifactId>rxkotlin</artifactId> <version>3.x.y</version> </dependency>

Gradle：

implementation("io.reactivex.rxjava3:rxkotlin:3.x.y")

2.3 RxKotlin 2.x（对应 RxJava 2.x）

Maven：

<dependency> <groupId>io.reactivex.rxjava2</groupId> <artifactId>rxkotlin</artifactId> <version>2.x.y</version> </dependency>

Gradle：

implementation("io.reactivex.rxjava2:rxkotlin:2.x.y")

2.4 RxKotlin 1.x（对应 RxJava 1.x）

Maven：

<dependency> <groupId>io.reactivex</groupId> <artifactId>rxkotlin</artifactId> <version>1.x.y</version> </dependency>

Gradle：

implementation("io.reactivex:rxkotlin:1.x.y")

2.5 通过 JitPack 构建

你也可以借助 JitPack 让 Gradle 或 Maven 直接基于本仓库的某个快照、分支或提交进行构建。例如，要基于 3.x 分支构建，Gradle 配置如下：

repositories { maven { url 'https://jitpack.io' } } dependencies { implementation 'com.github.ReactiveX:RxKotlin:3.x-SNAPSHOT' }

Maven 配置如下：

<repositories> <repository> <id>jitpack.io</id> <url>https://jitpack.io</url> </repository> </repositories> <dependency> <groupId>com.github.ReactiveX</groupId> <artifactId>RxKotlin</artifactId> <version>3.x-SNAPSHOT</version> </dependency>

从仓库的 gradle.properties 可以看到当前版本号为3.0.1，settings.gradle.kts 中rootProject.name="rxkotlin"，与上述坐标一致。

三、核心扩展函数全览

这是 RxKotlin 的核心价值所在：将 RxJava 的各类类型转换为 Kotlin 惯用 API。以下分类展示 README 中列出的扩展函数目录（实现代码分别位于 observable.kt 与 flowable.kt 等文件）。

3.1 集合 → Observable 的转换

3.2 Observable 高阶操作符

3.3 集合 → Flowable 的转换

3.4 Flowable 高阶操作符

3.5 Single、Maybe、Completable 相关

3.6 订阅（subscribeBy）与资源管理（addTo）

注：README 中Array<T>与Array<T>→toFlowable 的返回类型标注为Observable<Double>/Flowable<Double>，应为原文笔误。从源码 observable.kt 看，Array<T>.toObservable()实际返回Observable<T>；flowable.kt 中Array<T>.toFlowable()实际返回Flowable<T>。

四、核心实现源码解析

4.1 toObservable / toFlowable 的底层调用

在 observable.kt 中，数组类型的toObservable()均通过asIterable().toObservable()实现，最终委托给Observable.fromIterable(this)；Array<T>.toObservable()则直接调用Observable.fromArray(*this)。

IntProgression.toObservable()有一个值得注意的优化分支：

fun IntProgression.toObservable(): Observable<Int> = if (step == 1 && last.toLong() - first < Integer.MAX_VALUE) Observable.range(first, Math.max(0, last - first + 1)) else Observable.fromIterable(this)

即当步长为 1 且范围不会溢出Int时，走Observable.range的高效路径；否则回退为fromIterable。这组逻辑在 ObservableTest.kt 的intProgressionStep1、intProgressionDownTo、intProgressionOverflow用例中被验证。

Flowable侧（flowable.kt）逻辑与Observable完全对称，区别在于底层使用Flowable.fromArray、Flowable.range、Flowable.fromIterable。

4.2 subscribeBy 的具名参数实现

subscribers.kt 是subscribeBy系列方法的实现所在。其核心思想是为每个回调提供默认值占位（onNextStub、onErrorStub、onCompleteStub），并在封装为 RxJava 的Consumer/Action时，若用户未提供对应回调则替换为 RxJava 内部优化过的空实现：

private fun <T : Any> ((T) -> Unit).asConsumer(): Consumer<T> { return if (this === onNextStub) Functions.emptyConsumer() else Consumer(this) } private fun ((Throwable) -> Unit).asOnErrorConsumer(): Consumer<Throwable> { return if (this === onErrorStub) Functions.ON_ERROR_MISSING else Consumer(this) }

Completable.subscribeBy甚至根据是否提供回调选择subscribe()、subscribe(onComplete)或完整重载，以匹配 RxJava 中为 Completable 准备的优化消费者。测试 CompletableTest.kt、SingleTest.kt 中的testSubscribeByErrorIntrospection等用例，验证了未提供onError时返回的 Disposable 确实具备hasCustomOnError() == false的 LambdaConsumerIntrospection 行为。

4.3 Disposable 资源管理

disposable.kt 提供了两个资源管理扩展：

• CompositeDisposable.plusAssign(disposable)：运算符重载，支持composite += disposable的语法；
• Disposable.addTo(compositeDisposable)：将当前 Disposable 加入目标CompositeDisposable并返回自身，便于链式调用。

示例代码 examples.kt 中的addToCompositeSubscription展示了典型用法：

fun addToCompositeSubscription() { val compositeSubscription = CompositeDisposable() Observable.just("test") .delay(100, TimeUnit.MILLISECONDS) .subscribe() .addTo(compositeSubscription) compositeSubscription.dispose() }

4.4 高阶操作符的等价实现

README 表格中的多个“语法糖”操作符，本质上都是对 RxJava 既有操作符的薄封装，从源码看其对应关系为：

• mergeAll()≡flatMap { it }，concatAll()≡concatMap { it }，switchLatest()≡switchMap { it }（见 observable.kt、flowable.kt）；
• flatMapSequence { body }≡flatMap { body(it).toObservable() }（见 observable.kt）；
• Observable<Pair<A,B>>.toMap()≡toMap({ it.first }, { it.second })，toMultimap()同理（见 observable.kt）；
• mergeAllSingles()≡flatMapSingle { it }、mergeAllMaybes()≡flatMapMaybe { it }、mergeAllCompletables()≡flatMapCompletable { it }（见 single.kt、maybe.kt、completable.kt）。

4.5 具体化类型：cast / ofType

Observable<*>.cast<R>()与ofType<R>()利用 Kotlin 的inline + reified特性，免去了手动传Class的样板代码：

inline fun <reified R : Any> Observable<*>.cast(): Observable<R> = cast(R::class.java) inline fun <reified R : Any> Observable<*>.ofType(): Observable<R> = ofType(R::class.java)

对应的测试 ObservableTest.kt 中testCast、testCastWithWrongType、testOfType验证了转换成功、类型不匹配抛ClassCastException、以及ofType按目标类型过滤的行为。

五、SAM 辅助工厂（Kotlin 1.4 起已过时）

这些方法在新类型推断算法出现后已过时：Kotlin 1.4 起这些方法因新的类型推断算法而过时，并将在未来的某个 RxKotlin 版本中被移除。

历史上，为应对 Kotlin 使用 RxJava 时遇到的 SAM 歧义问题，RxKotlin 提供了一系列辅助工厂与扩展函数来规避受影响的操作符：

Observables.zip() Observables.combineLatest() Observable#zipWith() Observable#withLatestFrom() Flowables.zip() Flowables.combineLatest() Flowable#zipWith() Flowable#withLatestFrom() Singles.zip() Single#zipWith() Maybes.zip()

从源码看，这些辅助方法依旧保留在 Observables.kt、Flowables.kt、Singles.kt、Maybes.kt 中，但带 lambda 的版本全部标注了@Deprecated，并给出ReplaceWith迁移提示。例如：

@Deprecated("New type inference algorithm in Kotlin 1.4 makes this method obsolete. Method will be removed in future RxKotlin release.", replaceWith = ReplaceWith("Observable.zip(source1, source2, combineFunction)", "io.reactivex.Observable"), level = DeprecationLevel.WARNING) inline fun <T1 : Any, T2 : Any, R : Any> zip( source1: Observable<T1>, source2: Observable<T2>, crossinline combineFunction: (T1, T2) -> R ): Observable<R> = Observable.zip(source1, source2, BiFunction<T1, T2, R> { t1, t2 -> combineFunction(t1, t2) })

未废弃的重载则依然有价值：例如Observables.zip(source1, source2)直接发射Pair<T1,T2>，Observables.combineLatest(source1, source2, source3)发射Triple<T1,T2,T3>，这些重载在 ObservableTest.kt 的testPairZip、testTripleZip中被验证。Flowable侧的combineLatest/zip发射 Pair/Triple 的重载同样保留且未废弃（见 Flowables.kt）。

六、与其他 Rx 库配合使用

RxKotlin 可以与其他 Rx 与 Kotlin 库配合使用，例如 RxAndroid、RxBinding 以及 TornadoFX/RxKotlinFX。这些库与 RxKotlin 都是模块化的，RxKotlin 本质上只是针对 RxJava 的一组扩展函数，可以与这些库一起使用，且不应存在重叠或依赖问题。

仓库中提供的一个完整集成示例是 src/examples/kotlin/io/reactivex/rxjava3/kotlin/examples/retrofit/retrofit.kt：它使用 Retrofit 2.9.0 +RxJava3CallAdapterFactory+ Moshi，将 HTTP 接口直接声明为Observable<MavenSearchResponse>，然后借助flatMapIterable展平响应并订阅输出——这展示了 RxKotlin 在真实网络请求场景中的组合方式。该示例所需的依赖已声明在 build.gradle.kts 的examplesImplementation配置中。

七、贡献指南

7.1 当前状态

⚠️（2023-10-16）由于缺乏能够以合理方式处理贡献的开发者，我们目前不接受贡献。

7.2 特性取舍准则

由于 Kotlin 使得为任何东西实现扩展都变得很容易，本项目必须保守地界定哪些特性在范围之内。意图创造语法糖的尝试会快速退化为"语法糖精"（过度语法糖），这种个人偏好应留在个人领域而非开源库中。

以下是判断你的贡献是否可能在 RxKotlin 范围内的基本准则：

• 该特性是否已经存在于 RxJava 中？
  • 如果没有，请在 RxJava 中提议该操作符。
  • 如果有，Kotlin 能否进一步简化该操作符？
• 它是否显著减少了样板代码量？
• 它是否让现有操作符更易使用？
• RxJava 中不包含该特性，是因为 Java 语言的局限，还是因为刻意决定不包含它？

7.3 Kotlin Slack 频道

欢迎加入 Kotlin Slack 的 #rx 频道：https://kotlinlang.slack.com/messages/rx

八、延伸资源

• 《Learning RxJava》（Packt 出品）：第 12 章涵盖 RxKotlin 以及 Kotlin 惯用方式下的 RxJava。
• 《Reactive Programming in Kotlin》（Packt 出品）：主要聚焦 RxKotlin，以及用 Kotlin 学习响应式编程。

通过以上章节，你已经完整掌握了 RxKotlin 的定位、依赖引入方式、全部扩展函数清单及其源码级实现原理。接下来，可以对照仓库中 src/test 目录下的测试用例，自行编写并验证符合自己业务场景的 RxKotlin 代码。

• 异步编程
• 后端