首先导入依赖:
implementation 'io.reactivex.rxjava2:rxjava:2.2.9'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
Rxjava使用观察者模式实现异步操作,主要和Retrofit配合使用
RxJava 有四个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。
与传统观察者模式不同， RxJava 的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。
-
onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。 -
onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。 -
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。
public class MainActivity extends AppCompatActivity { Disposable mDisposable; protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); //建立连接 observable.subscribe(observer);//写法反了,可以理解为观察者订阅被观察者,被观察者(observable)也就是更新数据的一方,而观察者订阅了 是接受数据的一方, } //创建一个上游 被观察者 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.onComplete(); } }); //创建一个下游 观察者Observer Observer<Integer> observer = new Observer<Integer>() { @Override public void onSubscribe(Disposable d) { Log.d("输出:", "subscribe");//开始订阅 } @Override public void onNext(Integer value) { Log.d("输出:", "" + value);//订阅下一个 } @Override public void onError(Throwable e) { Log.d("输出:", "error");//订阅出错 } @Override public void onComplete() { Log.d("输出:", "complete");//订阅完成 } }; }
observable.subscribe(observer);//写法反了,可以理解为观察者订阅被观察者,被观察者(observable)也就是更新数据的一方,而观察者订阅了 是接受数据的一方,就好像读者(观察者)订阅期刊(被观察者)一样,期刊
一旦更新会通知读者
ObservableEmitter和Disposable.
1、ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件
但是，请注意，并不意味着你可以随意乱七八糟发射事件，需要满足一定的规则：
- 上游可以发送无限个onNext, 下游也可以接收无限个onNext.
- 当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
- 当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
- 上游可以不发送onComplete或onError.
- 最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
注: 关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则, **并不一定会导致程序崩溃. ** 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.
public class MainActivity extends AppCompatActivity { protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { Log.d("输出:", "console--> 1"); emitter.onNext(1); Log.d("输出:", "console--> 2"); emitter.onNext(2); Log.d("输出:", "console--> 3"); emitter.onNext(3); Log.d("输出:", "console--> complete"); emitter.onComplete(); Log.d("输出:", "console--> 4"); emitter.onNext(4); } }).subscribe(new Observer<Integer>() { private Disposable mDisposable; private int i; @Override public void onSubscribe(Disposable d) { Log.d("输出:", "subscribe"); mDisposable = d; } @Override public void onNext(Integer value) { Log.d("输出:", "onNext: " + value); i++; if (i == 2) { Log.d("输出:", "dispose"); mDisposable.dispose(); Log.d("输出:", "isDisposed : " + mDisposable.isDisposed()); } } @Override public void onError(Throwable e) { Log.d("输出:", "error"); } @Override public void onComplete() { Log.d("输出:", "complete"); } }); } }