[Lesson15] RxJava

在 gradle (Module) 层级的 dependencies 中内加入：

implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
implementation 'io.reactivex.rxjava3:rxjava:3.0.0'

Observable(被观察者)/Observer（观察者）
在RxJava中，有发送资料流程然後终止资料流程的Observable物件，以及订阅Observable的Observer物件。 Observer订阅的Observable发送值、错误、或完成信号後，Observer会接收到通知。

流程：

1. 创建Observable
   Observable类似於Iterable，对於给定序列，Observable遍历该序列并发送每项，一旦Observer订阅了Observable，Observable就开始发送资料。
2. 给Observable资料发送
   每当Observable发送资料时，它会通过该onNext()方法通知Observer。一旦Observable已经发送了所有的值，它终止於：

• onComplete操作成功时调用。
• onError抛出Exception调用。

3. 创建Observer
   Observable使用subscribe()操作符订阅Observers。一旦Observable订阅Observer，只要Observer发送了如下的内容，Observable就作出回应：

• onNext：Observable已经发送了一个值。
• onError：发生了错误。
• onComplete：Observable发送完所有的值。

4. 分配Observer给Observable
   Observer接收到Observable发送的资料後，Observer执行任务

private void myObservable() {
        Observable mObservable = new Observable() {
            @Override
            protected void subscribeActual(@NonNull Observer observer) {
                observer.onNext("hello world");
                observer.onComplete();
            }
        };

        Observer mObserver = new Observer() {
            // 这是新加入的方法，在订阅後发送数据之前，会首先调用
            // Disposable可用於取消订阅
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onNext(@NonNull Object o) {
                Log.e("onNext", String.valueOf(o));
            }

            @Override
            public void onError(@NonNull Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        };
        mObservable.subscribe(mObserver);
}

Flowable(被观察者)/Subscriber(观察者)
上游的被观察者会响应下游观察者的资料请求，下游调用request()告诉上游发送多少个资料。避免大量资料堆积在调用链上，使内存一直处於较低水平。

private void myFlowable() {
        Flowable.range(0,5).subscribe(new Subscriber<Integer>() {
            Subscription subscription;

            // 订阅後会先调用，相当於onStart()
            // Subscription可用於请求数据、取消订阅
            @Override
            public void onSubscribe(Subscription s) {
                Log.w("flow","onSubscribe start");
                subscription = s;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer integer) {
                Log.w("flow","onNext-->"+integer);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }

            @Override
            public void onComplete() {
                Log.w("flow","onComplete");
            }
        });
}

Single/SingleObserver
Single类似於Observable，不同的是，它总是只发送一个值，或者一个错误通知，而不是发送一系列的值，所以当你使用一个单一连续事件流，这样你可以使用Single。Single观察者只包含两个事件，一个是正常处理成功的onSuccess，另一个是处理失败的onError。因此，不同於Observable需要三个方法onNext，onError，onCompleted，订阅Single只需要两个方法：

• onSuccess - Single发送单个的值到这个方法
• onError - 如果无法发送需要的值，Single发送一个Throwable对象到这个方法
  Single只会调用这两个方法中的其中一个，而且只会调用一次，调用了任何一个方法之後，订阅关系中止。

private void mySingle() {
        // 被观察者
        Single<String> single = new Single<String>() {
            @Override
            protected void subscribeActual(@NonNull SingleObserver<? super String> observer) {
                observer.onSuccess("success");
            }
        };

        // 订阅观察者SingleObserver
        single.subscribe(new SingleObserver<String>() {

            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onSuccess(@NonNull String s) {
                Log.d("onSuccess", s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
            }
        });
}

Completable/CompletableObserver
如果你的观察者连onNext事件都不关心，可以使用Completable，因为Completable只有onComplete和onError两个事件

private void myCompletable() {
        Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(@NonNull CompletableEmitter emitter) throws Throwable {
                emitter.onComplete();
            }
        }).subscribe(new CompletableObserver() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onComplete() {
                Log.e("onComplete","onComplete");
            }

            @Override
            public void onError(@NonNull Throwable e) {
            }
        });
}

Maybe/MaybeObserver
如果你有一个需求是可能发送一个资料或者不会发送任何资料，这时候你就需要Maybe，它类似於Single和Completable的混合体。

Maybe可能会调用以下其中一种情况：

• onSuccess或者onError
• onComplete或者onError

private void myMaybe() {
        Maybe<String> maybe = Maybe.create(new MaybeOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull MaybeEmitter<String> emitter) throws Throwable {
                // 发送一个资料的情况或onError，不需再调用onComplete
                emitter.onSuccess("hello");

                // 不需要发送资料的情况或onError
                // emitter.onComplete();
            }
        });

        maybe.subscribe(new MaybeObserver<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onSuccess(@NonNull String s) {
                Log.d("onSuccess", s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });
}

谢谢大家愿意花时间阅读，小弟弟我在此鞠躬