如需下载源码，请访问
https://github.com/fengchuanfang/Rxjava2Tutorial
本篇文章对应的Demo类为：com.edward.edward.Rxjava2Demo.Rxjava2_2_Observable
文章原创，转载请注明出处：
Rxjava2入门教程二：Observable与Observer响应式编程在Rxjava2中的典型实现
运行源码，打开app，点击Demo图标，可进入以下页面
在RxJava中，函数响应式编程具体表现为一个观察者（Observer）订阅一个可观察对象（Observable），通过创建可观察对象发射数据流，经过一系列操作符（Operators）加工处理和线程调度器（Scheduler）在不同线程间的转发，最后由观察者接受并做出响应的一个过程
ObservableSource与Observer是RxJava2中最典型的一组观察者与可观察对象的组合，其他四组可以看做是这一组的改进版或者简化版。
Observable
抽象类Observable是接口ObservableSource下的一个抽象实现，我们可以通过Observable创建一个可观察对象发射数据流。
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Hello World");
emitter.onComplete();
}
});
调用Observable.create方法，创建一个可观察对象，并通过onNext发送一条数据“Hello World”，然后通过onComplete发送完成通知。
Observer
创建一个观察者Observer来接受并响应可观察对象发射的数据
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
System.out.println(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("接受完成");
}
};
在onNext方法中接收到可观察对象发射的数据"Hello World",并做出响应——打印到控制台。
Observer订阅Observable
observable.subscribe(observer);
通过subscribe方法，使Observer与Observable建立订阅关系，Observer与Observable便成为了一个整体，Observer便可对Observable中的行为作出响应。
Emitter/Observer
通过Observable.create创建可观察对象时，我们可以发现具体执行发射动作的是由接口ObservableEmitter的实例化对象完成的，而ObservableEmitter<T> 继承自 接口Emitter<T>，查看源码接口Emitter的具体代码如下：
public interface Emitter<T> {
//用来发送数据，可多次调用，每调用一次发送一条数据
void onNext(@NonNull T value);
//用来发送异常通知，只发送一次，若多次调用只发送第一条
void onError(@NonNull Throwable error);
//用来发送完成通知，只发送一次，若多次调用只发送第一条
void onComplete();
}
onNext：用来发送数据，可多次调用，每调用一次发送一条数据
onError：用来发送异常通知，只发送一次，若多次调用只发送第一条
onComplete：用来发送完成通知，只发送一次，若多次调用只发送第一条
onError与onComplete互斥，两个方法只能调用一个不能同时调用
数据在发送时，出现异常可以调用onError发送异常通知也可以不调用，因为其所在的方法subscribe会抛出异常，
若数据在全部发送完之后均正常，可以调用onComplete发送一条完成通知
接口Observer中的三个方法（onNext,onError,onComplete）正好与Emitter中的三个方法相对应，对于Emitter中对应方法发送的数据或通知进行响应。
步骤简化
去掉中间变量可以对之前的代码简化为以下形式：
public void demo2() {
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Hello World");
emitter.onComplete();
}
})
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
System.out.println(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("接受完成");
}
});
}
再应用Rxjava中强大的操作符，可以将代码简化成以下形式：
public void demo3() {
Observable.just("Hello World")
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
System.out.println(s);
}
});
}
再通过λ表达式，可进一步简化
public void demo3_1() {
Observable.just("Hello World").subscribe(System.out::println);
}
其中，just操作符可用来发送单条数据，数字，字符串，数组，对象，集合都可以当做单条数据发送。
Consumer可以看做是对观察者Observer功能单一化之后的产物——消费者，上例中的Consumer通过其函数accept只接收可观察对象发射的数据，不接收异常信息或完成信息。
如果想接收异常信息或完成信息可以用下面的代码：
public void demo4() {
Observable.just("Hello World")
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
System.out.println(s);
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
throwable.printStackTrace();
}
}, new Action() {
@Override
public void run() throws Exception {
System.out.println("接受完成");
}
});
}
第二个参数Consumer规定泛型<Throwable>通过函数accept接收异常信息。
第三个参数Action也是对观察者Observer功能单一化之后的产物--行动，通过函数run接收完成信息，作出响应行动。
发送数据序列
Observable可以发送单条数据也可以发送数据序列
通过最基础的方法发送数据序列：
public void demo5(final List<String> list) {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
try {
for (String str : list) {
emitter.onNext(str);
}
emitter.onComplete();
} catch (Exception e) {
emitter.onError(e);
}
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
System.out.println(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("接受完成");
}
});
}
在subscribe方法中，遍历集合list中的String元素，通过emitter.onNext(str)逐一发送；发送完成后通过emitter.onComplete()发送完成通知；如果发送过程中遇到异常，通过emitter.onError(e)发送异常信息。
Observer中通过onNext接收emitter发送的每一条信息并打印到控制台（emitter发送几次，Observer便接收几次），通过onError(Throwable e)接收异常信息，onComplete()接收完成信息。
同样可以通过操作符对其进行简化，如下;
public void demo6(final List<String> list) {
Observable
.fromIterable(list)
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
System.out.println(s);
}
});
}
再用λ表达式，进一步简化
public void demo6_1(final List<String> list) {
Observable.fromIterable(list).subscribe(System.out::println);
}
其中fromIterable操作符，可用来将一个可迭代对象中的元素逐一发送
Disposable
在之前的例子中，可以看到Observer接口中还有一个方法
public void onSubscribe(Disposable d) {
}
是在观察者Observer与可观察对象Observable，建立订阅关系后，回调这个方法，并且传过来一个Disposable类型的参数，可通过Disposable来控制Observer与Observable之间的订阅。
无论观察者Observer以何种方式订阅可观察对象Observable，都会生成一个Disposable，如下：
public void demo7(final List<String> list) {
Disposable disposable1 = Observable.just("Hello World")
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
System.out.println(s);
}
});
Disposable disposable2 = Observable
.fromIterable(list)
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
System.out.println(s);
}
});
}
查看Disposable接口的源码，如下：
public interface Disposable {
void dispose();
boolean isDisposed();
}
其中isDisposed()方法用来判断当前订阅是否失效，dispose()方法用来取消当前订阅。
只有当观察者Observer与可观察对象Observable之间建立订阅关系，并且订阅关系有效时，Observer才能对Observable进行响应。如果Observer在响应Observable的过程中，订阅关系被取消，则Observer无法对取消订阅关系之后Observable的行为作出响应。
运行下面的代码，当Observable接收到第5条数据时，取消订阅关系。
public void demo8() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 10; i++) {
System.out.println("发送" + i);
emitter.onNext(i);
}
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
private Disposable disposable;
@Override
public void onSubscribe(Disposable d) {
disposable = d;
}
@Override
public void onNext(Integer integer) {
System.out.println("接收" + integer);
if (integer > 4) disposable.dispose();
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("数据接受完成");
}
});
}
控制台日志如下：
I/System.out: 发送0
I/System.out: 接收0
I/System.out: 发送1
I/System.out: 接收1
I/System.out: 发送2
I/System.out: 接收2
I/System.out: 发送3
I/System.out: 接收3
I/System.out: 发送4
I/System.out: 接收4
I/System.out: 发送5
I/System.out: 接收5
I/System.out: 发送6
I/System.out: 发送7
I/System.out: 发送8
I/System.out: 发送9
可以发现取消订阅关系之前，Observable发送一条数据，Observe便接收一条，但是取消订阅关系之后，Observe将不再接收Observable发送的数据。
