Android rxjava2的disposable

rxjava+retrofit處理網絡請求

在使用rxjava+retrofit處理網絡請求的時候，一般會采用對觀察者進行封裝，實現代碼復用和拓展。可以參考我的這篇文章：rxjava2+retrofit封裝處理網絡請求全解析。一種可行的封裝如下：

• 基類observer

public abstract class BaseObserver<T> implements Observer<T> {

    protected String errMsg = "";
    protected Disposable disposable;

    @Override
    public void onSubscribe(Disposable d) {
        disposable = d;
    }

    @Override
    public void onNext(T t) {}

    @Override
    public void onError(Throwable e) {
        LogUtils.d("Subscriber onError", e.getMessage());
        if (!NetworkUtils.isConnected()) {
            errMsg = "網絡連接出錯,";
        } else if (e instanceof APIException) {
            APIException exception = (APIException) e;
            errMsg = exception.getMessage() + ", ";
        } else if (e instanceof HttpException) {
            errMsg = "網絡請求出錯,";
        } else if (e instanceof IOException) {
            errMsg = "網絡出錯,";
        }

        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    @Override
    public void onComplete() {
        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }
}

• 封裝請求（登錄為例） 這里userService是retrofit接口類

public void login(String phone, String password, BaseObserver<ResponseBean<UidBean>> observer) {
        userService.login(phone,password)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(observer);
    }

• 方法調用

APIUser.getInstance().login(phone, password, new BaseObserver<ResponseBean<UidBean>>() {
             @Override
             public void onNext(ResponseBean<UidBean> responseBean) {
                ToastUtils.showShort("登錄成功");
             }
         });

關於disposable

rxjava雖然好用，但是總所周知，容易遭層內存泄漏。也就說在訂閱了事件后沒有及時取閱，導致在activity或者fragment銷毀后仍然占用着內存，無法釋放。而disposable便是這個訂閱事件，可以用來取消訂閱。但是在什么時候取消訂閱呢？我知道有兩種方式:

• 使用CompositeDisposable

看源碼，CompositeDisposable的介紹很簡單

A disposable container that can hold onto multiple other disposables and offers O(1) add and removal complexity.

一個disposable的容器，可以容納多個disposable，添加和去除的復雜度為O(1)。 
這里需要注意的是在該類的addAll方法有這么一句注釋

Atomically adds the given array of Disposables to the container or disposes them all if the container has been disposed

也就是說，如果這個CompositeDisposable容器已經是處於dispose的狀態，那么所有加進來的disposable都會被自動切斷。

所以說可以創建一個BaseActivity，用CompositeDisposable來管理訂閱事件disposable，然后在acivity銷毀的時候，調用compositeDisposable.dispose()就可以切斷所有訂閱事件，防止內存泄漏。

static final class CreateEmitter<T> 
implements ObservableEmitter<T>, Disposable {

    private static final long serialVersionUID = -3434801548987643227L;

    final Observer<? super T> observer;

    CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onError(Throwable t) {
        if (t == null) {
            t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
        }
        if (!isDisposed()) {
            try {
                observer.onError(t);
            } finally {
                dispose();
            }
        } else {
            RxJavaPlugins.onError(t);
        }
    }

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

    @Override
    public void setDisposable(Disposable d) {
        DisposableHelper.set(this, d);
    }

    @Override
    public void setCancellable(Cancellable c) {
        setDisposable(new CancellableDisposable(c));
    }

    @Override
    public ObservableEmitter<T> serialize() {
        return new SerializedEmitter<T>(this);
    }

    @Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }

    @Override
    public boolean isDisposed() {
        return DisposableHelper.isDisposed(get());
    }
}

• 在oError和onComplete后調用disposable.dispose();，也就是上面我給的例子中的方法。

查看源碼，ObservableCreate的靜態類CreateEmitter就是這種方式實現的。同時也可以看到，onError和onComplete不可以同時調用的原因：每次掉用過onError或onComplete其中一個方法后，就會掉用dispose()方法，此時訂閱取消，自然也就不能掉用另一個方法了

除此之外，在github發現一個開源庫RxLifecyclee，粗略了解發現他實現的原理是綁定acvitvity是生命周期，在onStart中綁定就在onStop中解綁，其他onResume，onCreate同理。這個和第一種方式似乎又差不多，只不過第一種方式簡單，只在ondestory的時候銷毀所有事件。

所以那兩種方法哪種更好，我也不是很清楚。等到踩到什么坑了可能就知道了。 
如果某位大佬知道，希望不吝指教。