本文来说下关于Rxjava的几个问题

文章目录

概述
概念小贴士
知识要点
RxJava怎么通过被订阅者传给订阅者的过程是什么样的
Observer处理完onComplete后会还能onNext吗
RxJava中map、flatMap的区别，你还用过其他哪些操作符
Maybe、Observer、Single、Flowable、Completable几种观察者的区别，以及他们在什么场景用
RxJava切换线程是怎么回事
RxJava的subscribeOn只有第一次生效
RxJava的observeOn多次调用哪个有效
RxJava1.0、RxJava2.0、RxJava3.0有什么区别

概述

提升开发效率，降低维护成本一直是开发团队永恒不变的宗旨。当年Netflix为了增加服务器的性能和吞吐量，编写了RxJava并开源。很多人在深入学习和探索之后，感受到RxJava的巨大魅力！

官网：https://github.com/ReactiveX/RxJava

Android平台上已经有一些做异步操作的类库，但我们为什么还是要选择RxJava呢？简洁！有美感！RxJava的实现是链式调用，没有任何的嵌套，即使业务逻辑的越来越复杂，它依然能够保持简洁！仅仅是因为简洁吗？当然不！RxJava的操作符群非常强大！很多你需要的功能都已经实现了。

RxJava最大的作用是提供一个优秀的，现成的响应式/流式调用封装，而你只需付出些许学习成本就可以少做很多工作。

概念小贴士

大家都知道RxJava上手是非常难的一个框架，为什么说是难呢，因为它的功能非常强大，各种操作符让人很难上手，搭配使用带生命周期的框架有RxLife等。

以至于后面出了很多类似Rxjava的框架，有RxAndroid，我们用的RxJava切换主线程就是出自该框架，后面ACC架构中有LiveData、Lifecycle、包括后面协成中出来的flow异步流，都是向Rxjava靠拢，不过ACC出来的这些框架他们最大的特点就是操作简单，上手简单。

所以RxJava强大的框架背后读懂源码是非常难的，因此秉着扒开源码的想法，探索一些RxJava平时的疑惑。

在Rxjava里面有几个角色我们需要弄明白：

Observable：俗称被订阅者，被订阅者是事件的来源，接收订阅者(Observer)的订阅，然后通过发射器(Emitter)发射数据给订阅者。
Observer：俗称订阅者，注册过程传给被订阅者，订阅者监听开始订阅，监听订阅过程中会把Disposable传给订阅者，然后在被订阅者中的发射器(Emitter)发射数据给订阅者(Observer)。
Emitter：俗称发射器，在发射器中会接收下游的订阅者(Observer)，然后在发射器相应的方法把数据传给订阅者(Observer)。
Consumer：俗称消费器，这是RxJava2.0才出来的，在RxJava1.0中用Action来表示，消费器其实是Observer的一种变体，Observer的每一个方法都会对应一个Consumer，比如Observer的onNext、onError、onComplete、onSubscribe都会对应一个Consumer。
Disposable：俗称释放器，通常有两种方式会返回Disposable，一个是在Observer的onSubscribe方法回调回来，第二个是在subscribe订阅方法传consumer的时候会返回

这里以一种最基本的订阅来介绍它们之间的关系：

首先在①、②、③这几个过程中会形成一个从Observer到Observable订阅的过程，首先是下游的Observer发起Observable的subscribe方法，而该方法会调用到Observable的subscribeActual，并且会把下游的Observer传给该方法中，这就是上面的过程①。
过程②中会创建发射器，也就是各种Emitter对象，该发射器需要接受过程①中的Observer，也就是下游的观察者，接着给下游的Observer的添加订阅的监听，也就是onSubscribe方法，并且把Disposable传给onSubscribe方法，这里的Disposable其实是刚才创建的发射器，因为本身发射器也是实现了Disposable类型，所以下游的Observer在订阅监听中会收到上游创建的Disposable。
在过程②中做完了下游的Observer订阅监听后，继续向上一层的Observable添加订阅，也就是把②中创建的发射器或②中创建的Observer传给上游的Observable的subscribe方法。
完成了上面的从下游的observer到上游的Observable订阅的过程，接着就是④发射数据了，由于最上游的Observable会在subscribe方法中收到发射器，因此我们可以利用发射器把数据发送到下游的Observer，也就是onNext、onError、onComplete等方法。

知识要点

该篇文章会通过源码的形式介绍一下几点，以及总结面试过程中如何应答RxJava的问题。

RxJava怎么通过被订阅者传给订阅者的过程是什么样的?
Observer处理完onComplete后会还能onNext吗?
RxJava中map、flatMap的区别，你还用过其他哪些操作符?
Maybe、Single、Flowable、Completable几种观察者的区别，以及他们在什么场景用？
RxJava切换线程是怎么回事?
RxJava的subscribeOn只有第一次生效?
RxJava的observeOn多次调用哪个有效?
RxJava1.0、RxJava2.0、RxJava3.0有什么区别？

RxJava怎么通过被订阅者传给订阅者的过程是什么样的

通常我们写一个从订阅到发送数据的示例如下:

Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {emitter.onNext(1);emitter.onNext(2);emitter.onNext(3);emitter.onComplete();}
}).subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.d(TAG, "onSubscribe:" + d.getClass().getName());}@Overridepublic void onNext(@NonNull Integer integer) {Log.d(TAG, "onNext: " + integer);}@Overridepublic void onError(@NonNull Throwable e) {Log.d(TAG, "onError: " + e.getMessage());}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}
});

相信这是最简单的事件发送的示例，这没什么好说的，那它们是怎么发送数据，接收数据的呢，下面我会把代码拆分来看，因为现在是链式调用，我把代码拆分如下:

订阅过程

这里把创建observable和observer、以及发起订阅分别拆开来写，后面方便我们分析代码，首先是第一步发起订阅observable.subscribe(observer)：

@Override
public final void subscribe(@NonNull Observer<? super T> observer) {try {subscribeActual(observer);} catch (NullPointerException e) { // NOPMDthrow e;} catch (Throwable e) {NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");throw npe;}
}

observable的订阅方法关键一句subscribeActual(observer)，这里提一句，所有的被观察者的订阅入口都是subscribeActual方法，而subscribeActual在被观察者中是抽象方法，因此看对应的observable子类实现的逻辑，在上面通过Observable.create创建的被观察者是ObservableCreate，它是Observable的子类，我么需要明确，RxJava中的操作符都会对应一个Observable的子类，比如just操作符对应的是ObservableJust的被观察者，好了，我们看ObservableCreate的subscribeActual实现：

@Override
protected void subscribeActual(Observer<? super T> observer) {//创建发射器，并且把下游的observer给发射器CreateEmitter<T> parent = new CreateEmitter<>(observer);//给下游的observer添加被订阅的监听observer.onSubscribe(parent);try {//给上游的ObservableOnSubscribe添加订阅，并且把下游的observer给上游的ObservableOnSubscribesource.subscribe(parent);} catch (Throwable ex) {Exceptions.throwIfFatal(ex);parent.onError(ex);}
}

先是创建CreateEmitter类型的发射器，把下游的observer传给发射器，注意此处的发射器是实现了Disposable接口，所以紧接着会把发射器通过下游的观察者的onSubscribe方法传给下游观察者，注意此处传的是Disposable对象。接着会给上游的ObservableOnSubscribe添加订阅，并且把下游的observer给上游的ObservableOnSubscribe。

为了描述订阅的过程，我们画一张时序图：

订阅是从下游的Observer向上游的Observable发送订阅，然后在订阅的过程中，给下游的Observer发送订阅监听，并且给上游的被观察者添加订阅。

发送数据

上面我们知道在ObservableCreate的subscribeActual方法中给上游的ObservableOnSubscribe添加了onSubscribe订阅过程，并且把当前的发射器传给了ObservableOnSubscribe，而在我们上面的示例中定义的ObservableOnSubscribe内部类的subscribe方法通过传过来的发射器添加了如下代码:

emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();

所以到这里可以看到是通过发射器的onNext和onComplete发送数据，而emitter是上面订阅过程传过来的CreateEmitter，所以直接看它的onNext和onComplete：

@Override
public void onNext(T t) {if (t == null) {onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));return;}//如果isDisposed为false，则可以继续发送数据if (!isDisposed()) {observer.onNext(t);}
}

很简单，给observer发送数据，而当前的observer是订阅过程中传进来的下游observer，所以大家明白了吧，最终是下游的observer接收到数据。

发送主要通过上游的被观察者通知发射器，然后发射器会发送给下游的observer。

Observer处理完onComplete后会还能onNext吗

上面我们看到emitter.onNext三次完了后，会发送onComplete事件，那onComplete处理啥呢：

@Override
public void onComplete() {if (!isDisposed()) {try {observer.onComplete();} finally {dispose();}}
}

这是发射器中onComplete的定义，dispose方法是控制是否还能发送数据，其实这里的 CreateEmitter它是一个AtomicReference原子类包装Disposable的实现类，而我们dispose方法正是将该原子类添加了常量的DISPOSED，而在onNext方法中通过判断isDisposed是否为false才能继续发送数据。而isDisposed什么时候为false呢？

当AtomicReference中的包装对象不是DISPOSED。所以我们的onComplete是用来控制不能发送数据的。

您可以通过如下代码测试：

emitter.onNext(1);
emitter.onNext(2);
emitter.onComplete();
emitter.onNext(3);

看看下游的observer是否还能收到3的数据。

小总结

onComplete是用来控制不能发送数据的，也就是不能onNext了，包括onError也是不能再发送onNext数据了，该方法中也是调用了dispose方法。

RxJava中map、flatMap的区别，你还用过其他哪些操作符

map和flatMap是我们经常用的转换操作，我们先看看map如何使用：

Observable<Integer> createObservable = Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {emitter.onNext(1);emitter.onNext(2);emitter.onNext(3);emitter.onComplete();}});Observable<String> mapObservable = createObservable.map(new Function<Integer, String>() {@Overridepublic String apply(Integer integer) throws Throwable {return String.valueOf(integer + 1);}});Observer<String> observer = new Observer<String>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.d(TAG, "onSubscribe:" + d.getClass().getName());}@Overridepublic void onNext(@NonNull String string) {Log.d(TAG, "onNext: " + string);}@Overridepublic void onError(@NonNull Throwable e) {Log.d(TAG, "onError: " + e.getMessage());}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}};mapObservable.subscribe(observer);
}

通过createObservable的map操作生成了一个mapObservable的被观察者，最终通过mapObservable与observer形成订阅关系，而map操作需要一个Function的接口，第一个泛型是入参类型，第二个泛型是出参的类型，也就是apply的返回值，这里定义map的出参类型是String类型。我们再来看下flatMap如何使用：

Observable<Integer> flatMapObservable = mapObservable.flatMap(new Function<String, ObservableSource<Integer>>() {@Overridepublic ObservableSource<Integer> apply(String s) throws Throwable {return Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {emitter.onNext(Integer.valueOf(s)+1);emitter.onComplete();}});}
});
flatMapObservable.subscribe(observer);

在上面的mapObservable基础上通过flatMap返回flatMapObservable，最后通过flatMapObservable订阅observer。flatMap的Function第二个泛型是ObservableSource类型的，Observable的父类是ObservableSource类型，因此第二个参数返回Observable也可以。

从上面可以看出map是通过原始数据类型返回另外一种数据类型，而flatMap是通过原始数据类型返回另外一种被观察者。

关于面试也有问flatMap和concatMap的区别，下面我通过一个例子来演示他们的区别:

Observable<String> createObservable = Observable.just("1", "2", "3", "4", "5", "6", "7", "8", "9");
Observable<Integer> flatMapObservable = createObservable.flatMap(new Function<String, ObservableSource<Integer>>() {@Overridepublic ObservableSource<Integer> apply(String s) throws Throwable {if (s.equals("2")) {return Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {emitter.onNext(Integer.valueOf(s) + 1);emitter.onComplete();}}).delay(500, TimeUnit.MILLISECONDS);} else {return Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {emitter.onNext(Integer.valueOf(s) + 1);emitter.onComplete();}});}}
});
Observable<Integer> observeOnObservable = flatMapObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
Observer<Integer> observer = new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.d(TAG, "onSubscribe:" + d.getClass().getName());}@Overridepublic void onNext(@NonNull Integer string) {Log.d(TAG, "onNext: " + string);}@Overridepublic void onError(@NonNull Throwable e) {Log.d(TAG, "onError: " + e.getMessage());}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}
};
observeOnObservable.subscribe(observer);

在上面flatMap操作过程中为了演示flatMap和concatMap的区别，在数据为2的时候让返回的observable延迟500毫秒，我们看到的结果如下:

上面例子中3是由2的发射数据发射过来的，而正好数据为2的时候让延迟了500毫秒，那如果换成concatMap结果是按照发射数据的顺序来返回的。

concatMap和flatMap的功能是一样的，将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据放进一个单独的Observable。只不过最后合并ObservablesflatMap采用的merge，而concatMap采用的是连接(concat)。

总之一句一话,他们的区别在于：concatMap是有序的，flatMap是无序的，concatMap最终输出的顺序与原序列保持一致，而flatMap则不一定，有可能出现交错。

Maybe、Observer、Single、Flowable、Completable几种观察者的区别，以及他们在什么场景用

其实想知道它们的区别，我们直接看对应的Observer的方法有哪些:

Maybe

Maybe从字面意思是可能的意思，看下MaybeObserver接口：

public interface MaybeObserver<@NonNull T> {  void onSubscribe(@NonNull Disposable d);void onSuccess(@NonNull T t);  void onError(@NonNull Throwable e);void onComplete();
}

它没有onNext方法，也就是说不能发多条数据，如果回调到onSuccess再不能发消息了，如果直接回调onComplete相当于没发数据，也就是说Maybe可能不发送数据，如果发送数据只会发送单条数据。

Observer

这个不用多说了，它是能发送多条数据的，直到发送onError或onComplete才不会再发送数据了，当然它也是可以不发送数据的，直接发送onError或onComplete。

Single


public interface SingleObserver<@NonNull T> {void onSubscribe(@NonNull Disposable d);void onSuccess(@NonNull T t);void onError(@NonNull Throwable e);
}

single也是发送单条数据，但是它要么成功要么失败。

Flowable

Flowable没有FlowableObserver接口，它是由FlowableSubscriber代表观察者，Flowable在后面被压的时候讲，我们只要知道它是被压策略的一个被观察者。

Completable

public interface CompletableObserver {void onSubscribe(@NonNull Disposable d);void onComplete();void onError(@NonNull Throwable e);
}

Completable不发送数据，只会发送成功或失败的事件，当然这个用得很少。

小总结

从上面各个对应的observer接口来看，如果只想发一条数据，或者不发数据就用Maybe，如果想法多条数据或者不发数据就用Observable，如果只发一条数据或者失败就用Single，如果想用背压策略使用Flowable，如果不发数据就用Completable。

RxJava切换线程是怎么回事

大家都知道RxJava切换线程使用subscribeOn指定被观察者的在哪个线程执行，使用observeOn指定观察者在哪个线程执行，通常我们写法如下:

subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())

subscribeOn(Schedulers.io())

subscribeOn会返回一个ObservableSubscribeOn，它是一个Observable，根据前面介绍的订阅流程，我们直接看ObservableSubscribeOn的subscribeActual操作：


@Override
public void subscribeActual(final Observer<? super T> observer) {//创建了内部的Observer，其实这里类似上面介绍的Observable.create创建的发射器，只不过发射器是Emitterfinal SubscribeOnObserver<T> parent = new SubscribeOnObserver<>(observer);//给下游的observer添加订阅的监听observer.onSubscribe(parent);//给SubscribeOnObserver设置disposable对象parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

创建了SubscribeOnObserver对象，它是Observer类型的，其实类似上面介绍的Observable.create创建的发射器，只不过发射器是Emitter类型。接着给下游的observer添加订阅的监听，最后是给SubscribeOnObserver设置disposable对象，还记得在observable.create最后一步是给上游的ObservableOnSubscribe添加订阅吗，那我们看看此处是如果给上游的observable添加订阅的，首先scheduler是Schedulers.io()，最终它是一个IoScheduler对象，关于如何创建的线程池就不细说了，大家可以跟着IoScheduler创建过程可以看下，里面最终是通过SchedulerPoolFactory.create创建了线程池，创建线程池如下:

ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);

而scheduler.scheduleDirect(new SubscribeTask(parent))中的SubscribeTask是一个Runnable，所以最终通过线程池执行SubscribeTask的run方法：

到了最后还是通过线程池执行Runnable来添加上游Observable的订阅，并且把当前创建的SubscribeOnObserver传给了上游的observable，这个跟我们上面介绍Observable.create中给上游的ObservableOnSubscribe添加订阅是一样的。
小总结

subscribeOn实际是创建了ObservableSubscribeOn的Observable，它的订阅方法里面创建了SubscribeOnObserver，通过线程池执行Runnable来达到上游Observable的订阅在子线程中执行，这就是为什么subscribeOn能控制observable在哪个线程中执行的原因。

observeOn(AndroidSchedulers.mainThread())

同样如此observeOn也会有对应的observable，它是ObservableObserveOn，我们直接看它订阅的方法：

同样如此，可以看到先是拿到AndroidSchedulers中的worker，它是HandlerWorker类型，按道理说应该给下游的observer添加订阅监听啊，怎么没有呢，看官别急，我们继续看ObserveOnObserver的订阅方法：

我们的重点不在下游的observer订阅监听这，在ObserveOnObserver的onNext方法中，会调用schedule方法，最终是通过HandlerWorker的schedule执行ObserveOnObserver，因为ObserveOnObserver也是一个runnable实现类，HandlerWorker中的schedule方法是通过主线程的Handler给主线程发送了一个Message，所以我们回到ObserveOnObserver的run方法，在run方法中会执行下游的onNext、onError等方法，所以这就是为什么observeOn能让observer能在主线程中执行。

小总结

observeOn实际是创建了ObservableObserveOn的Observable，它的订阅方法里面创建了ObserveOnObserver，而ObserveOnObserver是实现了Runnable接口，把它包装成message给主线程的Handler发送一条消息，而ObserveOnObserver的run方法中会给下游的Observer发送数据。所以这就是observeOn能让observer在哪个线程中执行。

RxJava的subscribeOn只有第一次生效

如果你理解了订阅的过程，其实该问题很好理解，subscribeOn是规定上游的observable在哪个线程中执行，如果我们执行多次的subscribeOn的话，从下游的observer到上游的observable的订阅过程，最开始调用的subscribeOn返回的observable会把后面执行的subscribeOn返回的observable给覆盖了，因此我们感官的是只有第一次的subscribeOn能生效。

那如何才能知道它实际在里面生效了呢，我们可以通过doOnSubscribe来监听切实发生线程切换了。

RxJava的observeOn多次调用哪个有效

上面分析了observeOn是指定下游的observer在哪个线程中执行，所以这个更好理解，看observeOn下一个observer是哪一个，所以多次调用observeOn肯定是最后一个observeOn控制有效。

RxJava1.0、RxJava2.0、RxJava3.0有什么区别

RxJava2.0相比于RxJava1.0

添加背压的策略Flowable
添加Observer的变体consumer
ActionN 和 FuncN 改名（Action0 改名成Action，Action1改名成Consumer，而Action2改名成了BiConsumer，而Action3 - Action9都不再使用了，ActionN变成了Consumer<Object[]> 。Func改名成Function，Func2改名成BiFunction，Func3 - Func9 改名成 Function3 - Function9，FuncN 由 Function<Object[], R> 取代。）
Observable.OnSubscribe 变成 ObservableOnSubscribe
ObservableOnSubscribe 中使用 ObservableEmitter 发射数据给 Observer，在RxJava中使用Subscriber发射数据。
Subscription 改名为 Disposable

RxJava3.0相比与RxJava2.0

提供Java 8 lambda友好的API
删除Maybe.toSingle(T)
删除Flowable.subscribe(4 args)
删除Observable.subscribe(4 args)
删除Single.toCompletable()

你以为完了吗，还有背压没介绍呢，好吧，由于篇幅原因，把背压放到下一篇单独来讲了。

该篇介绍了Observable的订阅过程，其实我们只要把订阅过程理解为当前的Observable向下游的Observer添加订阅监听，向上游的Observable添加订阅，发送数据的是从上游的Observable向下游的Observer发送数据。

介绍了发射器发送了onComplete事件后再不能发送onNext事件的原因，介绍了map、flatMap以及concatMap的区别，以及简单的介绍了其他的关键字。

介绍了Maybe、Single、Flowable、Completable几种观察者的区别，以及他们在什么场景用。

介绍了RxJava切换线程通过subscribeOn控制上游的Observable订阅发生的线程，observeOn控制下游的observer接收数据发生的线程，以及为什么RxJava的subscribeOn只有第一次生效，RxJava的observeOn多次调用离observer最近的一个才生效。

介绍了RxJava2.0相比于RxJava1.0有哪些变动，RxJava3.0相比与RxJava2.0有哪些变动。

关于Rxjava的几个问题相关推荐

RxJava 实现模糊搜索
实现的效果图如下下面说下实现的具体方法 1 引入库 implementation "io.reactivex.rxjava3:rxjava:3.0.0-RC5"implement ...
RxJava firstElement 与 lastElement 以及 elementAt
1 firstElement 文档如下 2 lastElement 文档如下 3 elementAt 文档如下下面写一个下代码 firstElement Observable.just(1,2,3, ...
RxJava 过滤操作符 throttleFirst 与 throttleLast 以及 sample
看文档发现 throttleFirst 与 throttleLast 以及 Sample 都跳到同一个界面Sample throttleFirst :在某段时间内,只发送该段时间内第1次事件(假如一个 ...
RxJava 过滤操作符 distinct 和 distinctUntilChanged
distinct 看下文档 distinct : 过滤掉重复的元素 distinctUntilChanged: 过滤掉连续重复的元素,不连续重复的是不过滤看下代码 1 distinct Obse ...
RxJava 过滤操作符 take 与 takeLast
take 看下官方文档 take : 指定观察者正序接受指定的items数量 takeLast 指定观察者正序接受最后指定的items的数量看下demo take的代码 Observable.ju ...
RxJava 过滤操作符skip 与 skipLast
skip 看下文档 skip 是正序跳过指定的items skipLast 是正序跳过指定最后几个items 下面看下代码 Observable.just(1,2,3,4,5,6).skip(1)// ...
RxJava 变换操作符Map
看下文档如下通过对每个项目应用函数来转换Observable发出的项目个人理解为转换类型下面写一个把int 类型转换为String 类型的demo Observable.create(new O ...
RxJava 操作符 do
看下文档给的图片注册一项操作以应对各种可观察的生命周期事件 do的操作符有很多具体如下下面看下器使用 Observable.create(new ObservableOnSubscribe< ...
RxJava debounce()和throttleWithTimeout()
官方地址:http://reactivex.io/documentation/operators/debounce.html debounce :防抖动 throttleWithTimeout:节流超 ...
RxJava 解除订阅---------Disposable.dispose()方法
有时候我们需要解绑订阅,或者取消订阅, 这个时候就使用到了 Disposable.dispose()方法下面以一个案例说下使用方法 //Disposable.dispose()切断观察者与被观察者 ...

关于Rxjava的几个问题