Rxjava2 介绍与详解实例

目录

• 前言
• RX介绍
• Rx模式
• Rx使用依赖：
• Rxjava的入门基础 1. Observable
• 2. Flowable
• 3. Single
• 4. Completable
• 5. Maybe
• 6. Subject
• 7. Scheduler

前言

现在我们可以看到越来越多的开发者都在使用 Rx 相关的技术进行 App，Java 后端等领域进行开发。在开源的社区以及互联网公司，Rx、响应式编程、函数式都是热门的存在。所以笔者将结合自身的学习以及实际使用情况，写一个针对 Rxjava2 的系列文章，一起学习和使用 Rxjava 所带来的便捷。

笔者将利用工作之余，结合 ReactiveX 官方

Wiki

相关参考链接：

• 官网
• Wiki
• GitHub
• Api-Doc

Rxjava2 系列文章目录：

• Rxjava2 介绍与详解实例
  
• Rxjava2 Observable的创建详解及实例
  
• Rxjava2 Observable的数据变换详解及实例（一）
  
• Rxjava2 Observable的数据变换详解及实例（二）
  
• Rxjava2 Observable的数据过滤详解及实例（一）
  
• Rxjava2 Observable的数据过滤详解及实例（二）
  
• Rxjava2 Observable的结合操作详解及实例
  
• Rxjava2 可连接的Observable（ConnectableObservable）操作详解及实例
  
• Rxjava2 Observable的错误处理操作详解及实例
  
• Rxjava2 Observable的辅助操作详解及实例（一）
  
• Rxjava2 Observable的辅助操作详解及实例（二）
  
• Rxjava2 Observable的条件操作符详解及实例
  
• Rxjava2 Observable的布尔操作符详解及实例
  

• 未完，待续（正在努力整理中）...

实例代码：

RX介绍

ReactiveX的历史

ReactiveX

什么是ReactiveX

微软给的定义是，Rx是一个函数库，让开发者可以利用可观察序列和LINQ风格查询操作符来编写异步和基于事件的程序，使用Rx，开发者可以用Observables表示异步数据流，用LINQ操作符查询异步数据流， 用Schedulers参数化异步数据流的并发处理，Rx可以这样定义：Rx = Observables + LINQ + Schedulers。

ReactiveX.io给的定义是，Rx是一个使用可观察数据流进行

异步编程

观察者模式

迭代器模式

函数式编程

RxJava 到底是什么

RxJava 在 GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。这就是 RxJava ，概括得非常精准。

然而，对于初学者来说，这还是比较含蓄难懂的。因为它是一个

总结

异步

观察者模式

RxJava 优点

同样是做异步，为什么去使用它，而不用现成的 Thread，ThreadPoolExecutor，Android的AsyncTask / Handler / ... ？其实就是

简洁，易用

异步操作很关键的一点是程序的简洁性，因为在调度过程比较复杂的情况下，异步代码经常会既难写也难被读懂。 正如Android 创造的 AsyncTask 和Handler ，其实都是为了让异步代码更加简洁。RxJava 的优势也是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁。

名词定义

• Reactive 直译为反应性的，有活性的，根据上下文一般翻译为反应式、响应式。
• Iterable 可迭代对象，支持以迭代器的形式遍历，许多语言中都存在这个概念。
• Observable 可观察对象，在Rx中定义为更强大的Iterable，在观察者模式中是被观察的对象，一旦数据产生或发生变化，会通过某种方式通知观察者或订阅者。
• Observer 观察者对象，监听Observable发射的数据并做出响应，Subscriber是它的一个特殊实现。
• emit 直译为发射，发布，发出，含义是Observable在数据产生或变化时发送通知给Observer，调用Observer对应的方法，文章里一律译为发射。
• items 直译为项目，条目，在Rx里是指Observable发射的数据项，文章里一律译为数据，数据项。

Rx模式

使用观察者模式

• 创建：Rx可以方便的创建事件流和数据流
• 组合：Rx使用查询式的操作符组合和变换数据流
• 监听：Rx可以订阅任何可观察的数据流并执行操作

简化代码

• 函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态
• 简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码
• 异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制
• 轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种
  并发问题

使用Observable的优势

Rx扩展了观察者模式用于支持数据和事件序列，添加了一些操作符，它让你可以声明式的组合这些序列，而无需关注底层的实现：如线程、同步、线程安全、并发数据结构和非阻塞IO。

Observable通过使用最佳的方式访问异步数据序列填补了这个间隙。

Rx的Observable模型让你可以像使用集合数据一样操作异步事件流，对异步事件流使用各种
简单、可组合的操作。

1. Observable可组合

对于单层的异步操作来说，Java中Future对象的处理方式是非常简单有效的，但是一旦涉及到嵌套，它们就开始变得异常繁琐和复杂。使用Future很难很好的组合带条件的异步执行流程（考虑到运行时各种潜在的问题，甚至可以说是不可能的），当然，要想实现还是可以做到的，但是非常困难，或许你可以用 Future.get() ，但这样做，异步执行的优势就完全没有了。从另一方面说，Rx的bservable一开始就是为组合异步数据流准备的。

2. Observable更灵活

Rx的Observable不仅支持处理单独的标量值（就像Future可以做的），也支持数据序列，甚至是无穷的数据流。 Observable 是一个抽象概念，适用于任何场景。Observable拥有它的近亲Iterable的全部优雅与灵活。

Observable是异步的双向push，Iterable是同步的单向pull，对比：

3. Observable无偏见

Rx对于对于并发性或异步性没有任何特殊的偏好，Observable可以用任何方式实现，线程池、事件循环、非阻塞IO、Actor模式，任何满足你的需求的，你擅长或偏好的方式都可以。无论你选择怎样实现它，无论底层实现是阻塞的还是非阻塞的，客户端代码将所有与Observable的交互都当做是异步的。

Rx使用依赖：

下列是笔者使用的版本（可根据实际情况进行选择）：

1. 使用Gradle依赖：

   implementation "io.reactivex.rxjava2:rxjava:2.2.12"

2. 使用Maven依赖或者Jar包下载 ：Rxjava 2.2.12
3. 其他版本以及相关下载 ：Maven

Rxjava的入门基础

1. Observable

1.1 观察者模式

基本概念：

Observable

Observer

subscribe

事件

在RxJava中，一个实现了 Observer 接口的对象可以订阅 (subscribe) 一个 Observable 类的实例。订阅者(subscriber) 对 Observable 发射 (emit) 的任何数据或数据序列作出响应。这种模式 简化了并发操作，因为它不需要阻塞等待 Observable 发射数据，而是创建了一个处于待命状态的观察者哨兵，哨兵在未来某个时刻响应Observable的通知。

RxJava 的事件回调方法:

onSubscribe()

onNext()

onCompleted()

onError()

• onSubscribe(): 当被观察者被观察者订阅的时候触发。
• onNext(): 当被观察者发送数据的时候通过此方法通知观察者数据变换。
• onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
• onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。

注意： 在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

1.2 Consumer 和 Action

这两个词意思分别是

消费者

行为

简单示例：

// 1. 进行订阅，subscribe(Observer)
observable.subscribe(observer);

System.out.println("---------------------------------------------");
// 2. 进行订阅，subscribe(Consumer onNext)
observable.subscribe(nextConsumer);

System.out.println("---------------------------------------------");
// 3. 进行订阅，subscribe(Consumer onNext, Consumer onError)
observable.subscribe(nextConsumer, errorConsumer);

System.out.println("---------------------------------------------");
// 4. 进行订阅，subscribe(Consumer onNext, Consumer onError, Action onCompleted)
observable.subscribe(nextConsumer, errorConsumer, completedAction);

System.out.println("---------------------------------------------");
// 5. 进行订阅，subscribe(Consumer onNext, Consumer onError, Action onCompleted, Consumer onSubscribe)
observable.subscribe(nextConsumer, errorConsumer, completedAction, onSubscribeComsumer);

1.3 Observable的分类

在RxJava中，Observable 有 Hot 与 Cold 之分。

• Hot Observable : 无论有没有观察者进行订阅，事件始终都会发生。当有多个观察者订阅时，Hot Observable此时与订阅者们的关系时一对多的关系，可以与多个订阅者共享信息。
• Cold Observable : 只有有观察者订阅了，才开始执行数据流的发送，并且与观察者时一对一的关系。当有多个不同的订阅者时，消息是重新完整发送的，也就是说对于订阅者们来说，它们的事件是彼此独立的。

2. Flowable

Rxjava2.x 中有这么一个被观察者

Flowable

异步场景

3. Single

Single

因此，不同于Observable需要三个方法 onNext, onError, onCompleted，订阅Single只需要两个方法：

• onSuccess: Single发射单个的值到这个方法
• onError: 如果无法发射需要的值，Single发射一个Throwable对象到这个方法

Single 只会调用这两个方法中的一个，而且只会调用一次，调用了任何一个方法之后，订阅关系终止。

示例代码:

// Single: 只发送 onSuccess or onError 通知,并且只会发送一次, 第一次发送数据后的都不会在处理
Single.create(new SingleOnSubscribe<String>() {

@Override
public void subscribe(SingleEmitter<String> emitter) throws Exception {
emitter.onSuccess("Success");           // 发送success通知
emitter.onSuccess("Success2");      // 只能发送一次通知,后续不在处理
}
}).subscribe(new BiConsumer<String, Throwable>() {

@Override
public void accept(String t1, Throwable t2) throws Exception {
System.out.println("--> accept: t1 = " + t1 + ",  t2 = " + t2);
}
});

输出:

--> accept: t1 = Success,  t2 = null

toXXX

4. Completable

Completable 在创建后，不会发射任何数据, 只有

onComplete

onError

andThen

示例代码：

// 1. Completable：只发送complete 或 error 事件,不发送任何数据
Completable.fromAction(new Action() {

@Override
public void run() throws Exception {
System.out.println("Hello World! This is Completable.");
}
}).subscribe(new CompletableObserver() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError");
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});

System.out.println("----------------------------------------------");
// 2. 与 andThen 结合使用，当Completable执行完onCompleted后，执行andThen里的任务
Completable.create(new CompletableOnSubscribe() {

@Override
public void subscribe(CompletableEmitter emitter) throws Exception {
Thread.sleep(1000);
System.out.println("--> completed");
emitter.onComplete();
}
}).andThen(Observable.range(1, 5)).subscribe(new Consumer<Integer>() {

@Override
public void accept(Integer t) throws Exception {
System.out.println("--> accept: " + t);
}
});

输出：

--> onSubscribe
Hello World! This is Completable.
--> onComplete
----------------------------------------------
--> completed
--> accept: 1
--> accept: 2
--> accept: 3
--> accept: 4
--> accept: 5

toXXX

5. Maybe

Maybe 是 Rxjava 2.x 以后的新类型，只能发射 0 或者 1 项数据，即使后续有多个数据，后面的数据也不会被处理。可以看做是 Single 与 Completable 结合。

示例代码：

// Maybe 只发送0个或者1个数据，后续数据将被忽略
Maybe.create(new MaybeOnSubscribe<String>() {

@Override
public void subscribe(MaybeEmitter<String> emitter) throws Exception {
// 如果先发送了，将会调用MaybeObserver的onCompleted方法，如果有数据发送或者调用onError，则不会去调用
// emitter.onComplete();
emitter.onSuccess("Hello"); // 如果发送了第一个数据后续数据将不会被处理
emitter.onSuccess("World");
}
}).subscribe(new MaybeObserver<String>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}

@Override
public void onSuccess(String t) {
System.out.println("--> onSuccess: " + t);
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});

输出：

--> onSubscribe
--> onSuccess: Hello

toXXX

6. Subject

Subject

Observer

Observable

它既可以是数据源observerable，也可以是数据的订阅者Observer。这个可以通过源码来了解一下。

public abstract class Subject<T> extends Observable<T> implements Observer<T> {
...
}

Subject 实际上还是 Observable，不过它因为实现了Observer接口，可以通过onNext、onComplete、onError方法发射和终止发射数据。

注意： 不要使用

just(T)

from(T)

create(T)

在 Rxjava 中，官方一共为我们提供了几种Subject：

• AsyncSubject （仅释放接收到的最后一个数据）
• BehaviorSubject （释放订阅前最后一个数据和订阅后接收到的所有数据）
• PublishSubject （释放订阅后接收到的数据）
• ReplaySubject （释放接收到的所有数据）
• UnicastSubject (仅支持订阅一次的Subject)
• Serialized（串行化）
• TestSubject（在2.x中被TestScheduler和TestObserver替代）

6.1 AsyncSubject

AsyncSubject

可以获取数据业务逻辑的最后的结果数据。

注意： 如果因异常（Error）终止，将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。

实例代码：

// 注意: 不要使用just(T)、from(T)、create(T)来使用Subject，因为会把Subject转换为Obserable
// 无论订阅的时候AsyncSubject是否Completed，均可以收到最后一个值的回调
AsyncSubject<String> asyncSubject = AsyncSubject.create();
asyncSubject.onNext("emitter 1");
asyncSubject.onNext("emitter 2");
asyncSubject.onNext("emitter 3");
asyncSubject.onNext("emitter 4");
asyncSubject.onNext("emitter 5"); // 此时订阅后将近发送此项数据
// asyncSubject.onNext(1/0 + ""); // 发生error时将不会有数据发射，仅发送error通知
asyncSubject.onComplete();

// 订阅后只会接收最后一个数据
asyncSubject.subscribe(new Observer<String>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}

@Override
public void onNext(String t) {
System.out.println("--> onNext = " + t);
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError = " + e);
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});

输出：

--> onSubscribe
--> onNext = emitter 5
--> onComplete

6.2 BehaviorSubject

当观察者订阅

BehaviorSubject

可以缓存订阅前最后一次发出的数据，以及订阅后发送的所有数据。

注意： 如果因异常（Error）终止，将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。

实例代码：

// 创建无默认值的BehaviorSubject
BehaviorSubject<Integer> subject = BehaviorSubject.create();
// 创建有默认值的BehaviorSubject
BehaviorSubject<Integer> subjectDefault = BehaviorSubject.createDefault(-1);

// 观察者对象
Observer<Integer> observer = new Observer<Integer>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--------------------------------");
System.out.println("--> onSubscribe");
}

@Override
public void onNext(Integer t) {
System.out.println("--> onNext: " + t);
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
};

// 1. 无数据发送的时候，发送默认值
//  subjectDefault.subscribe(observer);

// 2. 此时会发射所有订阅后正常发射的数据: 1, 2, 3, 4, error
//  subject.subscribe(observer);
subject.onNext(1);
subject.onNext(2);
subject.onNext(3);

// 3. 此时会发射订阅前的一个数据及后面正常发射的数据: 3, 4, error
//  subject.subscribe(observer);
subject.onNext(4);
subject.onError(new NullPointerException());

// 4. 此时不会发射后续数据，仅发送Error通知
//  subject.subscribe(observer);
subject.onNext(5);
subject.onComplete();

// 5. 此时没有数据发射，如果有error存在的话，将会发送error
subject.subscribe(observer);

输出：

--------------------------------
--> onSubscribe
--> onNext: -1
--------------------------------
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onError: java.lang.NullPointerException
--------------------------------
--> onSubscribe
--> onNext: 3
--> onNext: 4
--> onError: java.lang.NullPointerException
--------------------------------
--> onSubscribe
--> onError: java.lang.NullPointerException
--------------------------------
--> onSubscribe
--> onError: java.lang.NullPointerException

6.3 PublishSubject

PublishSubject

ReplaySubject

如果 PublishSubject 在订阅前已经调用了

onComplete()

注意： 如果因异常（Error）终止，将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。

实例代码：

// 释放订阅后接收到正常发射的数据，有error将不会发射任何数据
PublishSubject<Integer> subject = PublishSubject.create();
// 观察者对象
Observer<Integer> observer = new Observer<Integer>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--------------------------------");
System.out.println("--> onSubscribe");
}

@Override
public void onNext(Integer t) {
System.out.println("--> onNext: " + t);
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
};

// 1. 此时订阅将释放后续正常发射的数据： 1，2, 3, 4, error
// subject.subscribe(observer);
subject.onNext(1);
subject.onNext(2);

// 2. 此时订阅，发射后续正常发射的数据：3, 4, error
// subject.subscribe(observer);
subject.onNext(3);
subject.onNext(4);

// 此时将不会发送任何数据，直接发送error
subject.onError(new NullPointerException());
subject.onNext(5);
subject.onComplete();

// 3. 此时订阅如果有error，仅发送error，否则无数据发射
subject.subscribe(observer);

输出：

--------------------------------
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onError: java.lang.NullPointerException
--------------------------------
--> onSubscribe
--> onNext: 3
--> onNext: 4
--> onError: java.lang.NullPointerException
--------------------------------
--> onSubscribe
--> onError: java.lang.NullPointerException

6.4 ReplaySubject

ReplaySubject

如果你把

ReplaySubject

ReplaySubject 还可以限制缓存数据的数量，限制缓存的时间：

• create(bufferAge)：指定内部缓存，减少内部缓存区增长过多的重分配
• createWithSize(maxAge)：指定订阅后接受之前已经发射过数据的 maxAge 个数据项
• createWithTime(timeout, TimeUnit, Scheduler) ：接受订阅后接受之前已经发射过指定 timeout 时间段内的数据项

实例代码：

// 1. 接受收到的所有数据以及通知，对每隔Observer都执行相同的独立的操作
ReplaySubject<Integer> subject = ReplaySubject.create();

// 2. 指定内部缓存大小，此方法避免在内部缓冲区增长以容纳新缓冲区时过多的数组重分配
// ReplaySubject<Integer> subject = ReplaySubject.create(5);

// 3. createWithSize(count)
// 指定保留订阅前数据项的个数的Subject，会发射订阅前count个数据和后续的数据
// ReplaySubject<Integer> subject = ReplaySubject.createWithSize(1);

// 4. createWithTime(maxAge, unit, scheduler)
// 指定保留订阅前指定maxAge时间段内数据和后续的数据
// ReplaySubject<Integer> subject = ReplaySubject.createWithTime(1, TimeUnit.MILLISECONDS, Schedulers.trampoline());

// 创建Observer(观察者), 可以接受Observable所有通知
Observer<Integer> observer = new Observer<Integer>() {

public void onSubscribe(Disposable d) {
System.out.println("----------------------------------");
System.out.println("--> onSubscribe");
}

public void onNext(Integer t) {
System.out.println("--> onNext = " + t);
}

public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

public void onComplete() {
System.out.println("--> onComplete");
}
};

// 正常接受所有Observable的数据和通知
subject.subscribe(observer);
subject.onNext(1);
subject.onNext(2);
subject.onNext(3);

// 正常接受所有Observable的数据和通知
subject.subscribe(observer);
subject.onNext(4);
// 如果有error，则发送error通知，不影响任何一个观察者数据与通知接受
// subject.onError(new NullPointerException());
subject.onNext(5);
subject.onComplete();

// 正常接受所有Observable的数据和通知
subject.subscribe(observer);

输出：

----------------------------------
--> onSubscribe
--> onNext = 1
--> onNext = 2
--> onNext = 3
----------------------------------
--> onSubscribe
--> onNext = 1
--> onNext = 2
--> onNext = 3
--> onNext = 4
--> onNext = 4
--> onNext = 5
--> onNext = 5
--> onComplete
--> onComplete
----------------------------------
--> onSubscribe
--> onNext = 1
--> onNext = 2
--> onNext = 3
--> onNext = 4
--> onNext = 5
--> onComplete

6.5 UnicastSubject

UnicastSubject

IllegalStateException

常用于一次性消费或安全场合，如网络结算，支付等。

实例代码：

// 创建UnicastSubject，只能被订阅一次，不能再次被订阅
UnicastSubject<Integer> subject = UnicastSubject.create();

// 创建Observer(观察者), 可以接受Observable所有通知
Observer<Integer> observer = new Observer<Integer>() {

public void onSubscribe(Disposable d) {
System.out.println("--------------------------------");
System.out.println("--> onSubscribe");
}

public void onNext(Integer t) {
System.out.println("--> onNext = " + t);
}

public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

public void onComplete() {
System.out.println("--> onComplete");
}
};
// 订阅后，此subject将不可以再被订阅了
subject.subscribe(observer);

subject.onNext(1);
subject.onNext(2);
subject.onNext(3);
// 此时会有IllegalStateException，因为只能订阅一次，不能重复订阅
subject.subscribe(observer);
subject.onNext(4);
subject.onNext(5);
subject.onComplete();

// 此时会有IllegalStateException，因为只能被订阅一次，不能重复订阅
subject.subscribe(observer);

输出：

--------------------------------
--> onSubscribe
--> onNext = 1
--> onNext = 2
--> onNext = 3
--------------------------------
--> onSubscribe
--> onError: java.lang.IllegalStateException: Only a single observer allowed.
--> onNext = 4
--> onNext = 5
--> onComplete
--------------------------------
--> onSubscribe
--> onError: java.lang.IllegalStateException: Only a single observer allowed.

6.6 SerializedSubject

在并发情况下，不推荐使用通常的Subject对象，此时会产生多次调用产生一系列不可控的问题，而是推荐使用

SerializedSubject

注意： 在Rxjava2 中

SerializedSubject

实例代码：

// 创建Subject
ReplaySubject<String> subject = ReplaySubject.create();

// 通过toSerialized()进行串行化
Subject<String> serialized = subject.toSerialized();

// 订阅
serialized.subscribe(new Consumer<String>() {

@Override
public void accept(String t) throws Exception {
System.out.println("--> accept: " + t + ", ReceiverThreadID: " + Thread.currentThread().getId());
}
});

// 多线程执行
for (int i = 0; i < 10; i++) {
final int value = i + 1;
new Thread(new Runnable() {

@Override
public void run() {
serialized.onNext(value + "-SendThreadID: " + Thread.currentThread().getId());
}
}).start();
}

System.in.read();

System.out.println("---------------------------------------------------------------------");

// 创建一个 SerializedObserver来进行串行化，保证线程安全
// 注意：只保证同时只有一个线程调用 onNext, onCompleted, onError方法，并不是将所有emit的值放到一个线程上然后处理
SerializedObserver<String> observer = new SerializedObserver<String>(new Observer<String>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe");
}

@Override
public void onNext(String t) {
System.out.println("--> onNext: " + t + ", ReceiverThreadID: " + Thread.currentThread().getId());
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError");
}

@Override
public void onComplete() {
System.out.println("--> onComplete");
}
});

// 订阅
subject.subscribe(observer);

// 多线程执行
for (int i = 0; i < 10; i++) {
final int value = i + 1;
new Thread(new Runnable() {

@Override
public void run() {
subject.onNext(value + "-SendThreadID: " + Thread.currentThread().getId());
//      if (value == 10) {
//          subject.onComplete();
//      }
}
}).start();
}

System.in.read();

输出：

--> accept: 1-SendThreadID: 11, ReceiverThreadID: 11
--> accept: 2-SendThreadID: 12, ReceiverThreadID: 11
--> accept: 10-SendThreadID: 20, ReceiverThreadID: 11
--> accept: 9-SendThreadID: 19, ReceiverThreadID: 11
--> accept: 8-SendThreadID: 18, ReceiverThreadID: 11
--> accept: 7-SendThreadID: 17, ReceiverThreadID: 11
--> accept: 6-SendThreadID: 16, ReceiverThreadID: 11
--> accept: 4-SendThreadID: 14, ReceiverThreadID: 11
--> accept: 5-SendThreadID: 15, ReceiverThreadID: 11
--> accept: 3-SendThreadID: 13, ReceiverThreadID: 11
---------------------------------------------------------------------
--> onSubscribe
--> onNext: 1-SendThreadID: 11, ReceiverThreadID: 11
--> onNext: 3-SendThreadID: 13, ReceiverThreadID: 11
--> onNext: 4-SendThreadID: 14, ReceiverThreadID: 11
--> onNext: 5-SendThreadID: 15, ReceiverThreadID: 11
--> onNext: 6-SendThreadID: 16, ReceiverThreadID: 16
--> onNext: 7-SendThreadID: 17, ReceiverThreadID: 16
--> onNext: 8-SendThreadID: 18, ReceiverThreadID: 16
--> onNext: 9-SendThreadID: 19, ReceiverThreadID: 16
--> onNext: 10-SendThreadID: 20, ReceiverThreadID: 16

6.7 TestSubject

Rxjava2 中已经取消了TestSubject，使用

TestScheduler

TestObserver

TestObserver 是一个一个记录事件并允许对其进行断言的观察者，多用于测试场合。一般可以创建一个

TestObserver

test()

实例代码：

// Observable
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {

@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(100);
emitter.onError(new NullPointerException());
emitter.onComplete();
}
});

// 1. 创建TestObserver对象
TestObserver<Integer> testObserver = TestObserver.create(new Observer<Integer>() {

@Override
public void onSubscribe(Disposable d) {
System.out.println("--> onSubscribe:");
}

@Override
public void onNext(Integer t) {
System.out.println("--> onNext: " + t);
}

@Override
public void onError(Throwable e) {
System.out.println("--> onError: " + e);
}

@Override
public void onComplete() {
System.out.println("--> onComplete:");
}
});

observable.subscribe(testObserver);
try {
// 断言是否为收到订阅，但是没有事件发送
testObserver.assertEmpty();
// 断言是否收到onComplete()
testObserver.assertComplete();
// 断言没有数据100发送
testObserver.assertNever(100);
// 断言接收数据结果
testObserver.assertResult(1, 2, 3);
// 断言异常
testObserver.assertError(NullPointerException.class);
... 更多请参考Api
} catch (Error e) {
System.out.println("Error: " + e);
}

System.out.println("-----------------------------------------------");
// Subject
AsyncSubject<Object> subject = AsyncSubject.create();

// 2. 从Observable或者Subject中获取TestObserver对象
TestObserver<Integer> test = observable.test();
TestObserver<Object> test2 = subject.test();
System.out.println(test.values()); // received onNext values
try {
// 断言是否为收到订阅，但是没有事件发送
test.assertEmpty();
test2.assertEmpty();
// 断言是否收到onComplete()
test.assertComplete();
// 断言没有数据100发送
test.assertNever(100);
// 断言接收数据结果
test.assertResult(1, 2, 3);
// 断言异常
test.assertError(NullPointerException.class);
... 更多请参考Api
} catch (Error e) {
System.out.println("Error: " + e);
}

输出（当出现断言不匹配的情况，会有相应Error抛出）：

--> onSubscribe:
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 100
--> onError: java.lang.NullPointerException
Error: java.lang.AssertionError: Value counts differ; expected: 0 but was: 4 (latch = 0, values = 4, errors = 1, completions = 0)
-----------------------------------------------
[1, 2, 3, 100]
Error: java.lang.AssertionError: Value counts differ; expected: 0 but was: 4 (latch = 0, values = 4, errors = 1, completions = 0)

6.8 Processor

Process

Subject

7. Scheduler

如果你想给Observable操作符链添加多线程功能，你可以指定操作符（或者特定的 Observable）在特定的调度器(

Scheduler

某些ReactiveX的Observable操作符有一些变体，它们可以接受一个

Scheduler

使用ObserveOn和SubscribeOn操作符，你可以让Observable在一个特定的调度器上执行, ObserveOn指示一个Observable在一个特定的调度器上调用观察者的onNext, onError和 onCompleted方法，SubscribeOn更进一步，它指示Observable将全部的处理过程（包括发射数据和通知）放在特定的调度器上执行。

调度器的种类

下表展示了RxJava中可用的调度器种类：

关于Rxjava中的线程模型、线程转换操作、调度器的使用等后面会有专题文章来详细介绍。

小结：

本章主要介绍了Rxjava的概念与添加使用依赖、Rxjava中的观察者模式、Observable、Flowable、Subject，Schedule等基础对象的介绍与使用，应该可以对Rxjava的概念及基本对象有了基本的认识和了解，以及简单的上手使用。

有关Rxjava2 的其他相关部分内容，后续将有系列的文章来介绍，请关注上面的实时文章目录。