前言
关于RxJava的介绍这里就不多说了，网上有很多相关的资料。
但有一点需要说明一下，很多同学可能在网上找到很多RxJava 1.X的教程，那么1.X和2.X有什么区别？学习2.X前需不需要先学习1.X？
其实1.X和2.X有很大的改变，如果你已学习过1.X，那么恭喜你，你只需要看看2.X有什么更新就可以了。如果你没学习过1.X，那么也不需要担心，你可以直接跳过1.X，来学习2.X。
所以本教程是直接使用2.X，概念性的东西这里也不多说，本教程直接使用例子一步一步带大家入门。
先在Android Studio 项目添加Gradle配置：
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.1.3'
可能你想添加最新的版本，那在哪里可以找到最新的版本呢？答案在这里：
https://github.com/ReactiveX/RxAndroid
入门例子：
//create创建一个上游Observable（被观察者）
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射事件
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
Log.d(TAG, "Observable发出：Complete");
e.onComplete();
Log.d(TAG, "Observable发出：4");
e.onNext(4);//虽然无法接收事件，但发送事件还是继续的
}
});
//创建一个下游Observer（观察者）
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext收到：" + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "Observer complete");
}
};
//建立连接（订阅），开始发送事件
observable.subscribe(observer);
ObservableEmitter是发射器，就是用来发出事件的，它可以发出三种类型的事件。通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。
写成RxJava引以为傲的链式操作：
//create创建一个上游Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
Log.d(TAG, "Observable发出：Complete");
e.onComplete();
Log.d(TAG, "Observable发出：4");
e.onNext(4);//虽然无法接收事件，但发送事件还是继续的
}
}).subscribe(new Observer<Integer>() {
//创建一个下游Observer（观察者）
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext收到：" + integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "Observer complete");
}
});
1、上游可以发送无限个onNext，下游也可以接收无限个onNext。
2、 当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送，而下游收到onComplete事件之后将不再继续接收事件。
3、当上游发送了一个onError后，上游onError之后的事件将继续发送，而下游收到onError事件之后将不再继续接收事件。
4、上游可以不发送onComplete或onError。
5、最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError，也不能先发一个onComplete, 然后再发一个onError，反之亦然。
注：关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则, 并不一定会导致程序崩溃. 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.
入门例子：
//create创建一个上游 Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
Log.d(TAG, "Observable发出：4");
e.onNext(4);
Log.d(TAG, "Observable发出：Complete");
e.onComplete();
Log.d(TAG, "Observable发出：5");
e.onNext(5);//虽然无法接收事件，但发送事件还是继续的
}
}).subscribe(new Observer<Integer>() {
//创建一个下游 Observer（观察者）
private Disposable mDisposable;
private int i;
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "subscribe");
mDisposable=d;
}
@Override
public void onNext(@NonNull Integer integer) {
Log.d(TAG, "onNext收到：" + integer);
i++;
if (i==2){
Log.d(TAG, "onNext：dispose");
mDisposable.dispose();//取消订阅，不再接收事件
Log.d(TAG, "onNext isDisposed : " + mDisposable.isDisposed());
}
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "Observer complete");
}
});
下游调用dispose()，取消收收事件，但并不会导致上游不再继续发送事件, 上游可以继续发送剩余的事件。
subscribe()有多个重载：
public final Disposable subscribe();
public final Disposable subscribe(Consumer<? super T> onNext);
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError);
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete);
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe);
public final void subscribe(Observer<? super T> observer);
带有一个Consumer参数的方法，表示下游只关心onNext事件，其他的事件不管。
因此，如果只需要onNext事件可以这么写：
//create创建一个上游 Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
Log.d(TAG, "Observable发出：4");
e.onNext(4);
Log.d(TAG, "Observable发出：Complete");
e.onComplete();
Log.d(TAG, "Observable发出：5");
e.onNext(5);//虽然无法接收事件，但发送事件还是继续的
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "onNext收到：" + integer);
}
});
网友评论