引言
经过几年的发展，响应式编程已经是很流行了，在Android开发中的应用也非常的广泛，身为Android开发者，则是必须掌握的技术。
正文
网上已经有很多很多RxJava相关的文章，视频等等教程，但是说实话对于入门，或者新手来说，确实不好理解，上来就是各种，观察者、被观察者、订阅、发布等等概念，一遍看下来直接就晕了，就感觉RxJava很难，难理解，用的时候也只是依葫芦画瓢，晕乎乎的用着，然后就没有然后了。
这里我都不说那些概念，因为讲概念太抽象，难记住，更难理解。我们用另外一个视角来学习。因为RxJava 1.x的版本 官方已经停止更新了维护了，没有学习过也没有关系，RxJava 2.x是全新的，直接学习使用就好了。
首先假设我们在工厂里上班，工厂都会有流水线，产品经过流水线生产后来订单了销售出去。
这里假设工厂生产的是一种六边形的“Jerry帅气饼干”，上游是生产车间流水线的事件流，下游是订单产品的销售消费事件流。中间连接上下游关系的暂且叫做“Jerry帅气饼干生产消费订单管理系统”（不要脸，名字写这么长），为了下文方便抒写且用“生产订单管理系统”（PCMS）。以上图上下游对应的就是Obsersvable被观察者也是发布者，下游对应Observer观察者也是订阅者。使用RxJava代码表示上图就是：
public void test1() {
// 上游生成产品流水线
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "test1 ====== Observable: ------ onNext: Jerry");
emitter.onNext("Jerry");
Log.d(TAG, "test1 ====== Observable: ------ onNext: 就是");
emitter.onNext("就是");
Log.d(TAG, "test1 ====== Observable: ------ onNext: 帅");
emitter.onNext("帅");
Log.d(TAG, "test1 ====== Observable: ------ onNext: ！！！");
emitter.onNext("！！！");
Log.d(TAG, "test1 ====== Observable: ------ onComplete");
emitter.onComplete();
Log.d(TAG, "test1 ====== Observable: ------ onNext: Jerry帅炸天！！！");
emitter.onNext("Jerry帅炸天！！！");
}
});
// 下游订单产品销售
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "test1 ====== Observer: onSubscribe");
}
@Override
public void onNext(String value) {
Log.d(TAG, "test1 ====== Observer: onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "test1 ====== Observer: onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "test1 ====== Observer: onComplete");
}
};
// 连接上下游的订单管理系统
observable.subscribe(observer);
}
上述代码，上游生产车间流水线就是Observable，下游订单销售就是Observer，中间通过“生产订单管理系统”subscribe来将上下游连接起来。
运行后输出结果是：
从输出结果来看，当上游Observable发出一个生产的饼干产品事件，下游订单销售的Observer就销售一个饼干产品事件，而且当上游调用了onComplete方法后，上游的生产事件还是生产饼干事件（继续生产了“Jerry帅炸天”饼干事件），但是下游的订单销售却没有消费掉。也就是事件产生方调用onComplete方法后，之后的事件还会继续发送，但是事件接收方就不会接收了。
我们来看看Observable的subscribe方法的参数：ObservableEmitter，Emitter顾名思义是发射器的意思，ObservableEmitter接口继承自Emitter接口：
public interface Emitter<T> {
/**
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(T value);
/**
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(Throwable error);
/**
* Signal a completion.
*/
void onComplete();
}
接口定义很简单，就三个方法，onNext我们上门已经用过了，是用来发射发送事件的，onComplete是用来表示事件发送完了，后面如果有新的事件发送，下游接收者可以不用处理，onError方法看注释说是发送一个异常事件给下游接收者。到底是不是这样，我们来试试就晓得了。
public void test2() {
// 上游生成产品流水线
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "test2 ====== Observable: ------ onNext: Jerry");
emitter.onNext("Jerry");
Log.d(TAG, "test2 ====== Observable: ------ onNext: 就是");
emitter.onNext("就是");
Log.d(TAG, "test2 ====== Observable: ------ onNext: 帅");
emitter.onNext("帅");
Log.d(TAG, "test2 ====== Observable: ------ onNext: ！！！");
emitter.onNext("！！！");
Log.d(TAG, "test2 ====== Observable: ------ onError");
emitter.onError(new IllegalStateException("Jerry饼干烤焦了，卖出去会被打！"));
Log.d(TAG, "test2 ====== Observable: ------ onNext: Jerry帅炸天！！！");
emitter.onNext("Jerry帅炸天！！！");
}
});
// 下游订单产品销售
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "test2 ====== Observer: onSubscribe");
}
@Override
public void onNext(String value) {
Log.d(TAG, "test2 ====== Observer: onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "test2 ====== Observer: onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "test2 ====== Observer: onComplete");
}
};
// 连接上下游的订单管理系统
observable.subscribe(observer);
}
在上游生产饼干的时候就生产了一个“Jerry饼干烤焦了，卖出去会被打！”的错误饼干事件，下游订单销售的onError出错状态会消费这个事件。而上游在出错事件后发送的“Jerry帅炸天！！！”饼干事件，同样也只是把事件发送了处理，下游订单销售并没有接收处理这个事件。
运行后输出结果：
细心的小伙伴应该会发现，每次执行的时候都会先调用下游的onSubscribe方法，这个方法里有个参数Disposable（用完即可丢弃）意思可以理解成，将上下游的连接切断，让上游的生产的饼干不打包放入下游订单销售环节，实际开发中是有这种需求的，当发送事件出问题的时候就需要断开事件接收处理。不像最近的疫苗事件，一些不要脸的生物疫苗公司把生产不合格的疫苗上市销售，伤天害理，谋财害命。下面举个例子：
public void test3() {
// 上游生成产品流水线
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "test3 ====== Observable: ------ onNext: Jerry");
emitter.onNext("Jerry");
Log.d(TAG, "test3 ====== Observable: ------ onNext: 就是");
emitter.onNext("就是");
Log.d(TAG, "test3 ====== Observable: ------ onNext: 帅");
emitter.onNext("帅");
Log.d(TAG, "test3 ====== Observable: ------ onNext: ！！！");
emitter.onNext("！！！");
Log.d(TAG, "test3 ====== Observable: ------ onComplete");
emitter.onComplete();
Log.d(TAG, "test3 ====== Observable: ------ onNext: Jerry帅炸天！！！");
emitter.onNext("Jerry帅炸天！！！");
}
});
// 下游订单产品销售
Observer<String> observer = new Observer<String>() {
private Disposable mDisposable;
private int i;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "test3 ====== Observer: onSubscribe");
mDisposable = d;
}
@Override
public void onNext(String value) {
Log.d(TAG, "test3 ====== Observer: onNext: " + value);
i++;
// 第一个事件接收后，就断开上下游连接
if (i == 1) {
Log.d(TAG, "test3 ====== Observer: start disposable");
mDisposable.dispose();
Log.d(TAG, "test3 ====== Observer: isDisposable: " + mDisposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "test3 ====== Observer: onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.d(TAG, "test3 ====== Observer: onComplete");
}
};
// 连接上下游的订单管理系统
observable.subscribe(observer);
}
这里我们在下游订单销售的onNext方法中，当接收完第一个饼干事件后，就使用mDisposable.dispose()方法将上下游的连接断开了，断开后上游后续生产的饼干事件，下游就接收不到。
运行的结果：
上图中，也验证了我们的猜想，当使用dispose断开上下游连接后，下游就无法再继续接收事件了。
这一讲就先介绍这么多，这样的方式理解Observable和Observer以及订阅动作subscribe是不是容易多了，希望对你有所帮助，下一讲使用RxJava来切换变化饼干事件处理的线程（主线程、子线程）。
