Android RxLife 一款輕量級別的RxJava生命週期管理庫（二）

時間 2019-12-07

標籤 android rxlife 一款輕量級別的 rxjava 生命週期管理欄目 Android 简体版

原文原文鏈接

在Android RxLife 一款輕量級別的RxJava生命週期管理庫（一）一文中，咱們講解了lift,compose這兩個操做符的使用，及RxJava是如何形成內存泄漏的，RxLife又是如何解決RxJava內存泄漏的。在這再也不次進行講解，若是尚未看的，請回頭閱讀java

本文主要講解RxLife在最新版本1.0.4中的as操做符的使用及其做用。git

gradle依賴github

implementation 'com.rxjava.rxlife:rxlife:1.0.4'

源碼下載歡迎starsegmentfault

看過上篇文章的同窗知道，在使用lift,compose這兩個操做符時，咱們規定了下游除了subscribe操做符外不能有其它的操做符出現，而後這種規定不是代碼層面的規定，須要開發者在開發時額外注意，顯然這種方案是不完美的，那麼咱們能不能從代碼層面上去規定呢？此時就要請咱們主角出場了！框架

as操做符

咱們先來看看as操做符如何使用。(如下代碼默認運行在Activity中)ide

Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        .as(RxLife.as(this)) //this 爲LifecycleOwner對象
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong);
        });

這樣就能夠了？是的，就這麼簡單。那麼as 是如何規避下游不會出現除subscribe以外的操做符呢？咱們先來看看使用as操做符，返回的啥？gradle

ObservableLife<Long> observableLife = Observable
        .intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        .as(RxLife.asOnMain(this));

返回的是一個ObservableLife對象，這是什麼鬼？進去看看ui

/**
 * User: ljx
 * Date: 2019/4/18
 * Time: 18:40
 */
public class ObservableLife<T> extends RxSource<Observer<? super T>> {

    private Observable<T> upStream;

    ObservableLife(Observable<T> upStream, LifecycleOwner owner, Event event, boolean onMain) {
        super(owner, event, onMain);
        this.upStream = upStream;
    }

    @Override
    public final Disposable subscribe() {
        return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }

    public final Disposable subscribe(Consumer<? super T> onNext) {
        return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        return subscribe(onNext, onError, Functions.EMPTY_ACTION, Functions.emptyConsumer());
    }

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                      Action onComplete) {
        return subscribe(onNext, onError, onComplete, Functions.emptyConsumer());
    }

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
                                      Action onComplete, Consumer<? super Disposable> onSubscribe) {
        ObjectHelper.requireNonNull(onNext, "onNext is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

        LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);

        subscribe(ls);

        return ls;
    }


    @Override
    public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            observer = RxJavaPlugins.onSubscribe(upStream, observer);

            ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

            subscribeActual(observer);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Disposable has been set or not
            // can't call onSubscribe because the call might have set a Subscription already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }
    }

    private void subscribeActual(Observer<? super T> observer) {
        Observable<T> upStream = this.upStream;
        if (onMain) {
            upStream = upStream.observeOn(AndroidSchedulers.mainThread());
        }
        upStream.onTerminateDetach().subscribe(new LifeObserver<>(observer, owner, event));
    }
}

仔細的你會發現，ObservableLife類裏面就只有一系列的subscribe方法，並且跟Observable裏面的subscribe方法是一一對應的，功能也是同樣的。到這，咱們就明白了，原來as操做符返回的是一個自定義的對象，並且該對象只提供了一系列subscribe方法，這樣就規避了下游不會再出現其它的操做符，就避免內存泄漏等危險。this

到這，as操做符的原理咱們就講清楚了，咱們再來看看as操做符的其它用法。idea

指定生命週期關閉管道

Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        //指定在onStop關閉管道，也能夠指定其它生命週期，不指定默認在onDestroy關閉管道
        .as(RxLife.as(this, Event.ON_STOP)) 
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong + " Thread=" + Thread.currentThread());
        });

指定觀察者在主線程回調

Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        .as(RxLife.asOnMain(this)) //asOnMain即表明在主線程回調
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong + " Thread=" + Thread.currentThread());
        });
        
        //等價於
Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        .observeOn(AndroidSchedulers.mainThread())
        .as(RxLife.as(this))
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong + " Thread=" + Thread.currentThread());
        });

同時指定線程回調及生命週期

Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        //在主線程回調，並指定在onStop關閉管道，也能夠指定其它生命週期，不指定默認在onDestroy關閉管道
        .as(RxLife.asOnMain(this, Event.ON_STOP)) 
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong + " Thread=" + Thread.currentThread());
        });
        //等價於
Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)
        .observeOn(AndroidSchedulers.mainThread())
        .as(RxLife.as(this, Event.ON_STOP))
        .subscribe(aLong -> {
            Log.e("LJX", "accept=" + aLong + " Thread=" + Thread.currentThread());
        });

ok，as操做符用法就這麼多，若是有更好的idea，請告訴我。