前言：
Android技能树系列：
Android基础知识
Android技能树 — Android存储路径及IO操作小结
数据结构基础知识
算法基础知识
Rx系列相关
Android技能树 — Rxjava取消订阅小结(1):自带方式
Android技能树 — Rxjava取消订阅小结(2):RxLifeCycle
现在很多项目都在使用Rxjava了，对于RxJava的使用，估计都很熟悉了，但是很多人在使用RxJava的时候容易产生内存泄漏问题，比如我们在用RxJava配合Retrofit的时候，发出请求出去，拿到数据后我们可能会去刷新界面，但是如果这时候网络比较差，返回比较慢，而我们的Activity这时候关闭了，那RxJava当拿到返回的数据的时候去刷新界面就会报空指针异常了。所以我们当Activity关闭的时候，我们这时候如果RxJava还没执行完，我们应该取消订阅。
常用的主要三种方式：（按照⭐️推荐从低到高来介绍）
- 自带取消订阅方式(⭐️)
- RxLifeCycle(⭐️⭐️)
- AutoDispose(⭐️⭐️⭐️)
本文主要讲自带取消订阅方式。
1. 自带取消订阅方式：
在RxJava 1的时候我们知道在你用Observable执行时候会返回一个Subscription类：
Subscription subscription = Observable.xxx("yy").subscribe(.....);
复制代码然后我们只需要在我们界面的ondestory方法中对这个对象进行取消订阅操作就可以：
@Override
protected void onDestroy() {
if (subscription != null && !subscription.isUnsubscribed) {
subscription. unsubscribe();
}
super.onDestroy();
}
复制代码我们可以看到很简单，这样当我们Activity关闭的时候已经自动取消了订阅。
而RxJava2换了方式，但是基本方法是一模一样的,只是换成了Disposable：
如果我们使用的是Consumer，那和原来RxJava 1 是一样的操作：
Disposable disposable = Observable.just(1).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
}
});
//取消订阅
if(disposable != null && !disposable.isDisposed()){
disposable.dispose();
}
复制代码但是我们可能更多的是使用Observer等，那这时候subscrbe(observer)返回的是void,所以不能像上面一样操作，需要下面代码所示那样：
private Disposable disposable;
Observable.just(1).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
disposable = d;
}
@Override
public void onNext(Integer integer) {}
@Override
public void onError(Throwable e) {}
@Override
public void onComplete() {}
});
//然后在需要取消订阅的地方调用即可
if(disposable != null && !disposable.isDisposed()){
disposable.dispose();
}
复制代码和RxJava 1 最大的区别主要是获取这个取消订阅对象的地方不同，Disposable是在Observer里面的onSubscribe方法的参数拿到，然后我们可以定义一个临时变量进行赋值，然后在需要取消订阅的地方去调用即可。
但是很多人会说难道不能和RxJava 1 的方式差不多，因为很多项目已经按照RxJava 1 的方式来封装了进行相应的取消订阅代码，直接换成RxJava 2 方式变化不一样了，能不能变得和Rxjava 1 取消订阅方式差不多 。答案是当然可以。
我们可以使用DisposableObserver和subscribeWith二者结合来做的和Rxjava 1 一样的方式来取消订阅。
1.1 DisposableObserver
DisposableObserver 是一个抽象的 Observer, 它通过实现了 Disposable 接口允许异步取消。
/**
* An abstract {@link Observer} that allows asynchronous cancellation by implementing Disposable.
*
* @param <T> the received value type
*/
public abstract class DisposableObserver<T> implements Observer<T>, Disposable {
final AtomicReference<Disposable> s = new AtomicReference<Disposable>();
@Override
public final void onSubscribe(Disposable s) {
if (DisposableHelper.setOnce(this.s, s)) {
onStart();
}
}
/**
* Called once the single upstream Disposable is set via onSubscribe.
*/
protected void onStart() {
}
@Override
public final boolean isDisposed() {
return s.get() == DisposableHelper.DISPOSED;
}
@Override
public final void dispose() {
DisposableHelper.dispose(s);
}
}
复制代码我们可以看到，这个DisposableObserver即实现了Observer，又实现了Disposable接口。
PS : DisposableObserver源码里面有个AtomicReference，有些人也许不知道这个类，可以初步理解为加了锁，方便多线程操作。具体可以看文章Java之美[从菜鸟到高手演练]之atomic包的原理及分析
所以我们初步代码可以变为:
//比如这个是我们的Observer
DisposableObserver observer = new DisposableObserver() {
@Override
public void onNext(Object o) {}
@Override
public void onError(Throwable e) {}
@Override
public void onComplete() {}
};
//把我们的Observer对Observable进行订阅
Observable.just(1).subscribe(observer);
//然后在需要取消订阅的地方对这个observer进行取消订阅即可。
observer.dispose();
复制代码1.2 subscribeWith
public final <E extends Observer<? super T>> E subscribeWith(E observer) {
subscribe(observer);
return observer;
}
复制代码我们可以看到 subscribeWith订阅的源码是把Observer对象同时返回，正好配合上面的DisposableObserver:
DisposableObserver observer = Observable.just(1).subscribeWith(new DisposableObserver<Integer>() {
@Override
public void onNext(Integer integer) {}
@Override
public void onError(Throwable e) {}
@Override
public void onComplete() {}
});
//需要取消订阅的地方：
observer.disposable();
复制代码这下是不是和我们RxJava 1 里面的写法一模一样了。
1.3 CompositeDisposable
我在看很多一些开源项目中，有些人一个界面的上会有多个订阅（比如有多个网络接口请求）,这时候我们需要批量取消订阅，有些人会写一个ArrayList，然后把这些上面我们返回的DisposableObserver对象加入到ArrayList中，然后当我们的界面关闭的时候，再遍历ArrayList，把里面的元素取出来一个个取消订阅。实际上RxJava 2 中有替我们考虑到这个需求。那便是CompositeDisposable类。
CompositeDisposable compositeDisposable = new CompositeDisposable();
//批量添加
compositeDisposable.add(observer1);
compositeDisposable.add(observer2);
compositeDisposable.add(observer2);
//最后一次性全部取消订阅
compositeDisposable.dispose();
复制代码
2. 配合MVP做封装：
我们以Activity为例：
public abstract class BaseFrameActivity<P extends BasePresenter, M extends BaseModel> extends BaseActivity implements BaseView {
public P mPresenter;
public M mModel;
@Override
protected void onCreate(Bundle savedInstanceState) {
//根据传进来的第一个泛型参数实例化
mPresenter = TUtil.getT(this, 0);
//根据传进来的第二个泛型参数实例化
mModel = TUtil.getT(this, 1);
if (this instanceof BaseView && mPresenter != null) {
//实例化的presneter绑定View和model
mPresenter.attachVM(this, mModel);
}
super.onCreate(savedInstanceState);
}
@Override
protected void onDestroy() {
if (mPresenter != null) {
//当onDestory的时候调用presenter的解除View和model的绑定
mPresenter.detachVM();
}
super.onDestroy();
}
}
复制代码比如我们在BaseFrameActivity里面传入了p 和 m 的泛型，我们需要动态实例化，当然你也可以用Dagger2等，比如我们是用反射：
public class TUtil {
public static <T> T getT(Object o, int i) {
try {
/**
* getGenericSuperclass() : 获得带有泛型的父类
* ParameterizedType ： 参数化类型，即泛型
* getActualTypeArguments()[] : 获取参数化类型的数组，泛型可能有多个
*/
return ((Class<T>) ((ParameterizedType) (o.getClass()
.getGenericSuperclass())).getActualTypeArguments()[i])
.newInstance();
} catch (InstantiationException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (ClassCastException e) {
e.printStackTrace();
}
return null;
}
// 获得类名className对应的Class对象
public static Class<?> forName(String className) {
try {
return Class.forName(className);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
return null;
}
}
复制代码我们来看BasePresenter.java:
public abstract class BasePresenter<M, V> {
public M mModel;
public V mView;
public RxManager mRxManager = new RxManager();
public void attachVM(V v, M m) {
this.mModel = m;
this.mView = v;
}
public void detachVM() {
mRxManager.clear();
mView = null;
mModel = null;
mDialog = null;
}
}
复制代码我们把Observable等取消订阅操作放在了RxManager里面了：
public class RxManager {
private CompositeDisposable compositeDisposable = new CompositeDisposable();
public void add(Disposable d) {
compositeDisposable.add(d);
}
public void clear() {
compositeDisposable.dispose();
}
}
复制代码最终比如我们要用自己的Activity了：
- 只需要继承BaseFrameActivity，然后把要实例化的P和M对象传入：
public class SplashActivity
extends BaseFrameActivity<XPresenter, XModel>
implements XContract.View {}
复制代码直接就可以使用mPresenter执行相关操作，并且mPresenter实例化的时候也已经实例化一个RxManager实例对象。
- 假设我们用的是最原始的Observer来订阅：
public class XPresenter extends XContract.Presenter {
@Override
public void getXImage() {
mModel
.getXImage()
.subscribe(new Observer<SplashImgEntity>() {
@Override
public void onSubscribe(Disposable d) {
//自动就会把Disposable加入到RxManager中的CompositeDisposable 中。
mRxManager.add(d);
}
@Override
public void onNext(SplashImgEntity splashImgEntity) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
}
复制代码然后Activity销毁时候，会自己去帮你取消订阅。
总结：
emmmmmm…….请多多指教。
