RxJava(十)：Disposable和Transformer的使用Android技术专题分享

在RxJav1.x中，Subscription的接口可以用来取消订阅。

publicinterfaceSubscription{/***Stopsthereceiptofnotificationsonthe{@linkSubscriber}thatwasregisteredwhenthisSubscription*wasreceived.*

*Thisallowsderegisteringan{@linkSubscriber}beforeithasfinishedreceivingallevents(i.e.before*onCompletediscalled).*/voidunsubscribe();/***Indicateswhetherthis{@codeSubscription}iscurrentlyunsubscribed.**@return{@codetrue}ifthis{@codeSubscription}iscurrentlyunsubscribed,{@codefalse}otherwise*/booleanisUnsubscribed();}在Rx.Javal.x中，Observable.subscribe()方法会返回一个Subscription的对象。也就是说,我们每次订阅都会返回Subscription。Subscription只需调用unsubscribe就可以取消订阅。

Subscription对象是被观察者和订阅者之间的纽带。RxJava使用Subscription处理取消订阅时，会停止整个调用链。如果使用了一串很复杂的操作符，则调用unsubscribe()将会在它当前执行的地方终止，而不需要做任何额外的工作。

在Android开发中，我们会在Activity/Fragment的onDestroy里做一些释放资源的事情，如果使用了RxJaval.x，则可以用Subscription.isUnsubscribed()检查Subscription是否是unsubscribed。如果是unsubscribed，则调用Subscription.unsubscribe(),RxJava会取消订阅井通知给Subscriber，井允许垃圾回收机制释放对象，防止任何RxJava造成内存泄露。

在RxJava2.0之后，Subscription被改名为Disposable。RxJava2.x中由于己经存在名为org.reactivestreams.Subscription这个类（遵循ReactiveStreams标准），为了避免名字冲突，所以将原先的rx.Subscription改名为io.reactivex.disposables.Disposable

publicinterfaceDisposable{/***Disposetheresource,theoperationshouldbeidempotent.*/voiddispose();/***Returnstrueifthisresourcehasbeendisposed.*@returntrueifthisresourcehasbeendisposed*/booleanisDisposed();}Disposabledisposable=Observable.just("hello,rxjava").subscribe(newConsumer(){@Overridepublicvoidaccept(Strings)throwsException{Log.d(TAG,"Next:"+s);}});disposable.dispose();1.1CompositeDisposableRxJava1.x中有一个复合订阅（compositesubscription）的概念。在RxJava2.x中，RxJava也内置了一个类似复合订阅的容器CompositeDisposable，每当我们得到一个Disposable时，就调用CompositeDisposable.add()，将它添加到容器中，在退出的时候，调用CompositeDisposable.clear（）即可切断所有的事件。在Android中，我们经常会看到这样的用法，也就是会有一个针对Activity或者Fragment的CompositeDisposable，这样就可以在onDestroy或者其他合适的地方取消订阅。

在Android开发中，可以使用Disposable来管理一个订阅或者使用CompositeDisposable来管理多个订阅，防止由于没有及时取消，导致Activity/Fragment无法销毁而引起内存泄露。然而，也有一些比较成熟的库可以做这些事情。

importcom.trello.rxlifecycle2.components.support.RxAppCompatActivity;publicclassBackPressureActextendsRxAppCompatActivity{}当涉及绑定Observable到Activity或者Fragment生命周期时，要么指定Observable应终止的生命周期事件，要么让RxLifecycle库决定何时终止Observable序列。

默认情况下，RxLifecycle将在辅助生命周期事件中终止Observable，所以如果在Activity的onCreate()方法期间订阅了Observable，则RxLifecycle将在该Activity的onDestroy()方法中终止Observable序列。如果在Fragment的onAttach()方法中订阅，RxLifecycle将在onDetach()方法中终止该序列。

AutoDispose是Uber开源的库。它与RxLifecycle的区别是，不仅可以在Android平台上使用，还可以在Java（的企业级）平台上使用，适用的范围更加宽广。

现在除CompositeDisposable和RxLifecycle外，还多了一个选择AutoDispose。AutoDispose支持Kotlin、AndroidArchitectureComponents，井且AutoDispose可以与RxLifecycle进行相互操作。

Transformer是转换器的意思。早在RxJava1.x版本就己有了Observable.Transformer、Single.Transformer和Completable.Transformer。在RxJava2.x版本中有ObservableTransformer、SingleTransformer、CompletableTransformer、FlowableTransformer、MaybeTransformer。其中，FlowableTransformer和MaybeTransformer是新增。由于RxJava2.x将Observable拆分成了Observable和Flowable，所以有了FlowableTransformer。同样，Maybe也是RxJava2.x新增的一个类型，所以有MaybeTransformer

Transformer能够将一个Observable/Flowable/Single/Completable/Maybe对象转换成另一个Observable/Flowable/Single/Completable/Maybe对象，与调用一系列的内联操作符一模一样。

举个简单的例子，写一个transformer()方法将一个发射整数Observable转换为发射字符串的Observable

privatestaticObservableTransformertransformer(){returnnewObservableTransformer(){@OverridepublicObservableSourceapply(Observableupstream){returnupstream.map(newFunction(){@OverridepublicStringapply(Integerinteger)throwsException{returnString.valueOf(integer);}});}};}在使用transformer()方法，通过标准RxJava操作。

Observable.just(123,456).compose(transformer()).subscribe(newConsumer(){@Overridepublicvoidaccept(Strings)throwsException{Log.d(TAG,"Next:"+s);}});//执行结果Next:123Next:4563.2与compose操作符结合使用compose操作符能够从数据流中得到原始的被观察者。当创建被观察者时，compose操作符会立即执行，而不像其他的操作符需要在onNext()调用后才能执行

常用的场景

对于网络请求，我们经常会做如下操作来切换线程：

.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())做一个简单的封装：

objectRxJavaUtils{@JvmStaticfunobservableToMain():ObservableTransformer{returnObservableTransformer{upstream->upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())}}@JvmStaticfunflowableToMain():FlowableTransformer{returnFlowableTransformer{upstream->upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())}}}这段代码是用Kotlin写的,把一些工具类用Kotlin来编写，而且Kotlin的Lambda表达式也更为直观。

对于Flowable切换到主线程的操作，可以这样使用

.compose(RxJavaUtils.flowableToMain())RxLifecycle中的LifecycleTransformerRxLifecycle能够配合Android的生命周期，防止App内存泄漏，其中就使用了LifecycleTransformer。知乎也做了一个类似的RxLifecycle，能够做同样的事情。