• 示例工程
• 教程

本篇技术指南围绕 cw-omnibus 仓库中的 RxAndroid 示例工程展开，该示例是《The Busy Coder's Guide to Android Development》一书的配套代码，核心目标是演示"如何使用 RxAndroid 在主应用程序线程（main thread）上订阅事件"。读完本文，你将掌握Observable的创建、subscribeOn/observeOn的线程切换语义、AndroidSchedulers.mainThread()的底层原理，以及用Disposable管理订阅生命周期的完整实战方案。

示例工程定位：一个"主线程订阅"的最小可运行范式

原 README 对工程的定位非常凝练："This sample Android app demonstrates using RxAndroid to subscribe to events on the main application thread."（本示例 Android 应用演示如何使用 RxAndroid 在主应用程序线程上订阅事件。）这揭示了两层含义：

1. 数据源的生产（emission）发生在后台线程，避免阻塞主线程；
2. 订阅者（subscriber）的消费发生在主线程，从而可以安全地更新 UI。

整个工程是一个可独立编译运行的 Gradle Android 项目，文件布局如下（相对仓库根目录）：

核心原理：subscribeOn 与 observeOn 的分工，以及 AndroidSchedulers.mainThread()

Android 的主线程（又称 UI 线程）负责处理界面绘制与输入事件，任何耗时的 CPU/IO 操作放在主线程都会导致卡顿甚至 ANR；反过来，UI 控件又只能在主线程上安全更新。RxJava 用两个核心调度操作符来解决这对矛盾：

• subscribeOn(scheduler)：指定数据源（上游）运行所在的线程。在示例中，Schedulers.newThread()让"生产事件"的代码跑在一条全新的后台线程上；
• observeOn(scheduler)：指定下游消费运行所在的线程，它只影响调用点之后的链路。示例中的AndroidSchedulers.mainThread()让onNext/onComplete等事件最终投递到 Android 主线程。

AndroidSchedulers.mainThread()正是 RxAndroid 这个扩展库的核心价值：它是 RxJava 调度器（Scheduler）与 AndroidLooper（主消息循环）之间的桥梁。RxJava 本身并不感知 Android 的Handler/Looper机制，而 RxAndroid 通过io.reactivex.android.schedulers.AndroidSchedulers包提供了主线程调度器，使 RxJava 事件流能够平滑接入 Android 的主线程消息队列。这一点从 RxDemoFragment.java 的导入语句import io.reactivex.android.schedulers.AndroidSchedulers;可以直观印证。

数据流全链路解析：从 25 个单词到 RecyclerView

示例的数据流全部在RxDemoFragment.onCreate()中构建，完整代码如下（见 RxDemoFragment.java）：

Observable<String> observable = Observable .create(source()) .subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .doOnComplete(() -> { Toast.makeText(getActivity(), R.string.done, Toast.LENGTH_SHORT).show(); }); sub = observable.subscribe(s -> { adapter.add(s); });

数据源source()返回一个ObservableOnSubscribe<String>（见 RxDemoFragment.java）：

private ObservableOnSubscribe<String> source() { return emitter -> { for (String item : ITEMS) { emitter.onNext(item); SystemClock.sleep(400); } emitter.onComplete(); }; }

整条链路的执行语义可以概括为：

ITEMS(25 个单词) │ emitter.onNext(item) + SystemClock.sleep(400) ▼ subscribeOn(Schedulers.newThread()) ← 数据源在后台新线程运行，模拟耗时生产 ▼ observeOn(AndroidSchedulers.mainThread()) ← 下游事件切回 Android 主线程 ▼ doOnComplete(...) ← 流结束时在主线程弹 Toast ▼ subscribe(s -> adapter.add(s)) ← 主线程上把单词追加进 RecyclerView

几个值得深挖的实现细节：

• Observable.create()+ObservableOnSubscribe：这是 RxJava 2 中手动驱动事件流的推荐方式，通过Emitter（发射器）调用onNext/onComplete/onError。示例中的发射器每onNext一个单词就SystemClock.sleep(400)一次，共 25 个单词，模拟一个约 10 秒、逐条产出数据的"慢速数据源"；
• Schedulers.newThread()每订阅一次就新建一条线程，适合低频、短命的后台任务；本例中整个发射循环（含 10 秒休眠）都被移出主线程，正是它保证了界面在数据生产期间依然可响应；
• observeOn(AndroidSchedulers.mainThread())放置的位置：它位于subscribeOn之后，因此只影响其后的doOnComplete与subscribe消费者——即 Toast 与adapter.add()都会在 Android 主线程执行；
• doOnComplete是"副作用"操作符：不改变流本身，只在流完成时执行附加逻辑，这里用于在主线程弹出R.string.done（值为 "Done!"，见 strings.xml）提示订阅全部完成。

UI 装配与增量渲染：RecyclerView + notifyItemInserted

界面的装配分为两层。宿主 Activity RxDemo.java 在onCreate中检查android.R.id.content上是否已存在 Fragment，不存在则通过FragmentTransaction.add()挂载RxDemoFragment，避免配置变更时重复添加。

Fragment 的布局 main.xml 是一个占满全屏的 RecyclerView：

<android.support.v7.widget.RecyclerView xmlns:android="http://schemas.android.com/apk/res/android" android:id="@android:id/list" android:layout_width="match_parent" android:layout_height="match_parent"/>

在onViewCreated中（见 RxDemoFragment.java），为 RecyclerView 配置了LinearLayoutManager与DividerItemDecoration（竖向分割线），并挂上适配器。

真正体现"主线程订阅"价值的是适配器的增量追加逻辑（见 RxDemoFragment.java）：

private void add(String word) { words.add(word); notifyItemInserted(words.size() - 1); }

每收到一个onNext，适配器就把单词追加到ArrayList并通过notifyItemInserted通知 RecyclerView 插入一行。由于该回调经由AndroidSchedulers.mainThread()投递，notifyItemInserted必然发生在主线程，符合 RecyclerView 的线程约束；行视图复用android.R.layout.simple_list_item_1，由内部静态类RVArrayAdapter与RowHolder完成绑定。运行后即可观察到列表以约 400ms 的间隔逐行"长"出单词，直到全部 25 个单词展示完毕。

订阅生命周期管理：Disposable 与 setRetainInstance

RxJava 2 用Disposable取代了 RxJava 1 的Subscription，调用dispose()即可取消订阅并切断数据流。示例在 Fragment 销毁时做了严谨的清理（见 RxDemoFragment.java）：

@Override public void onDestroy() { if (sub != null && !sub.isDisposed()) { sub.dispose(); } super.onDestroy(); }

与之配合的是setRetainInstance(true)（见 RxDemoFragment.java）。两者组合的意图清晰：

• 旋转等配置变更时：Fragment 实例被保留，onDestroy不会被调用，订阅继续运行，已收到的数据保存在model列表中，避免重复订阅、重复拉流；
• Fragment 真正销毁时：dispose()切断订阅，防止后台线程继续发射、防止主线程回调访问已销毁的 UI，从根源上规避内存泄漏与对已销毁组件的非法操作。

这里也引出一个通用最佳实践：凡是在onCreate/onStart中发起的订阅，都应在对称的onDestroy/onStop中释放，尤其是长期运行或跨线程的流。

构建配置一览：依赖版本与编译选项

示例工程的模块级构建脚本 app/build.gradle 完整展示了依赖组合：

dependencies { implementation "com.android.support:recyclerview-v7:27.1.1" implementation "com.android.support:support-fragment:27.1.1" implementation 'io.reactivex.rxjava2:rxjava:2.1.7' implementation 'io.reactivex.rxjava2:rxandroid:2.0.1' }

要点如下：

工程级配置方面，build.gradle 声明使用 Android Gradle Plugin3.1.4并依赖google()与jcenter()仓库；gradle-wrapper.properties 指定 Gradle4.4；gradle.properties 中设置了org.gradle.jvmargs=-Xmx1536M以分配构建进程堆内存；AndroidManifest.xml 声明包名com.commonsware.android.rx，RxDemo为主启动 Activity。需要说明的是，该工程未包含可执行脚本gradlew，最便捷的验证方式是将其作为独立工程导入 Android Studio，由 IDE 依据上述配置完成构建与运行（若本机装有 Gradle 4.4+ 亦可直接构建）。

运行观察要点与同系列示例延伸

运行该示例时，建议重点观察三个现象：

1. 界面无卡顿：虽然数据源线程在 10 秒内持续sleep(400)，但由于生产逻辑在Schedulers.newThread()的后台线程上，主线程始终可以响应触摸与滚动；
2. 逐行增量渲染：约每 400ms 有一行单词插入列表底部，这是notifyItemInserted的动画效果，也直接证明了onNext已被调度回主线程；
3. 结束提示：25 个单词全部输出后触发doOnComplete，弹出 "Done!" Toast，说明onComplete同样在主线程被消费。

cw-omnibus 仓库的 Rx 目录 下还收录了同一系列的其他示例，可作为延伸阅读：用 RxJava 改造既有AsyncTask的 Rx/Simple、实现更复杂 Observable 链路的 Rx/Observable、在后台解析 XML 资源的 Rx/XML、结合 Retrofit 使用的 Rx/Retrofit，以及使用 Java 8 lambda 编写 RxJava 的 Rx/Lambda。它们与本文示例共同构成了一套完整的 RxJava/RxAndroid 上手路线：从"主线程订阅"这个最小范式出发，逐步延伸到网络请求、资源解析与更复杂的响应式链路组合。

• 示例工程
• 教程