RxJava2 中 doFinally 和 doAfterTerminate 的比较

简介: RxJava2 中 doFinally 和 doAfterTerminate 的比较

在 RxJava 中 doFinally 和 doAfterTerminate 这两个操作符很类似,都会在 Observable 的 onComplete 或 onError 调用之后进行调用。


使用了这两个操作符在 Observable 结束后,会调用 doFinally、doAfterTerminate 所提供的 Action。


这两个操作符虽然有一定的相似度,但他们依然有差别。并且两者在使用时,会存在调用的先后顺序。


doAfterTerminate



从 doAfterTerminate 操作符的源码来看

@CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final Observable<T> doAfterTerminate(Action onFinally) {
        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
        return doOnEach(Functions.emptyConsumer(), Functions.emptyConsumer(), Functions.EMPTY_ACTION, onFinally);
    }


它调用的是 doOnEach() 方法。


其实,doOnNext、doOnError、doOnComplete、doOnTerminate 等这些操作符也是调用 doOnEach() 方法。


doOnEach() 实际上调用的是 ObservableDoOnEach 类。( RxJavaPlugins.onAssembly 本身是一个 hook 方法,会返回一个 Observable 对象。)


doOnEach() 需要四个参数:onNext、onError、onComplete、onAfterTerminate。

doAfterTerminate 操作符的参数 onFinally 对应的是 onAfterTerminate。

@CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    private Observable<T> doOnEach(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Action onAfterTerminate) {
        ObjectHelper.requireNonNull(onNext, "onNext is null");
        ObjectHelper.requireNonNull(onError, "onError is null");
        ObjectHelper.requireNonNull(onComplete, "onComplete is null");
        ObjectHelper.requireNonNull(onAfterTerminate, "onAfterTerminate is null");
        return RxJavaPlugins.onAssembly(new ObservableDoOnEach<T>(this, onNext, onError, onComplete, onAfterTerminate));
    }


ObservableDoOnEach 继承了 Observable 类,并实现了它的抽象方法 subscribeActual()。该方法是 Observable 和 Observer 连接的纽带。其中,source 代表了被观察者 Observable 本身,而 DoOnEachObserver 是实际的观察者。

@Override
    public void subscribeActual(Observer<? super T> t) {
        source.subscribe(new DoOnEachObserver<T>(t, onNext, onError, onComplete, onAfterTerminate));
    }


从 DoOnEachObserver 中的 onError()、onComplete() 方法中可以看到 onAfterTerminate 是在 downstream.onError(t) 或者 downstream.onComplete() 之后,才执行 run()。这也符合我们最初对它的认识。

@Override
        public void onError(Throwable t) {
            if (done) {
                RxJavaPlugins.onError(t);
                return;
            }
            done = true;
            try {
                onError.accept(t);
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                t = new CompositeException(t, e);
            }
            downstream.onError(t);
            try {
                onAfterTerminate.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                RxJavaPlugins.onError(e);
            }
        }
        @Override
        public void onComplete() {
            if (done) {
                return;
            }
            try {
                onComplete.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return;
            }
            done = true;
            downstream.onComplete();
            try {
                onAfterTerminate.run();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                RxJavaPlugins.onError(e);
            }
        }


doFinally



doFinally 是 RxJava 在 2.0.1 版本新增的操作符。


doFinally 除了拥有 doAfterTerminate 的特性之外,还会在下游(downstream)取消时被调用。这是 doFinally 和 doAfterTerminate 最大的区别。


同样,看一下 doFinally 的源码。它调用的是 ObservableDoFinally 类。

public final Observable<T> doFinally(Action onFinally) {
        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
        return RxJavaPlugins.onAssembly(new ObservableDoFinally<T>(this, onFinally));
    }


在 ObservableDoFinally 类中,从 subscribeActual() 可以看出它的实际观察者是 DoFinallyObserver 类。其中,onFinally 是 doFinally 操作符所传递的参数。

@Override
    protected void subscribeActual(Observer<? super T> observer) {
        source.subscribe(new DoFinallyObserver<T>(observer, onFinally));
    }


在 DoFinallyObserver 类中的 onError、onComplete、dispose 方法中都会调用 runFinally() 方法。而 runFinally() 执行的正是 onFinally 的 run()。

@Override
        public void onError(Throwable t) {
            downstream.onError(t);
            runFinally();
        }
        @Override
        public void onComplete() {
            downstream.onComplete();
            runFinally();
        }
        @Override
        public void dispose() {
            upstream.dispose();
            runFinally();
        }
        ......
        void runFinally() {
            if (compareAndSet(0, 1)) {
                try {
                    onFinally.run();
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    RxJavaPlugins.onError(ex);
                }
            }
        }


所以,从源码可以得出在 doFinally 是在观察者执行完 onError、onComplete 或取消时执行的操作。


二者的顺序



doFinally 和 doAfterTerminate 都会在 onComplete 之后才执行,那么它们二者的顺序是如何呢?


不妨写一段代码测试一下:

Observable.just("Hello World")
                .doOnNext(s -> System.out.println("doOnNext:"))
                .doFinally(() -> System.out.println("doFinally:"))
                .doAfterTerminate(() -> System.out.println("doAfterTerminate:"))
                .subscribe(
                        s -> System.out.println("onNext:" + s),
                        throwable -> System.out.println("onError:"),
                        () -> System.out.println("onComplete:"));


执行结果:

doOnNext:
onNext:Hello World
onComplete:
doAfterTerminate:
doFinally:


发现 doFinally 是在 doAfterTerminate 之后调用。


那么交换一下它们的顺序会如何呢?

Observable.just("Hello World")
                .doOnNext(s -> System.out.println("doOnNext:"))
                .doAfterTerminate(() -> System.out.println("doAfterTerminate:"))
                .doFinally(() -> System.out.println("doFinally:"))
                .subscribe(
                        s -> System.out.println("onNext:" + s),
                        throwable -> System.out.println("onError:"),
                        () -> System.out.println("onComplete:"));


执行结果:

doOnNext:
onNext:Hello World
onComplete:
doFinally:
doAfterTerminate:


这一次,doFinally 先执行而 doAfterTerminate 后执行。


因为,它们都需要在 downstream.onComplete() 执行之后,才会执行。而 downstream 对应的下游是观察者。下流的数据流向跟上游的数据流向是相反的,从下向上的。所以,离观察者越近,就越先执行。这就是两段代码执行顺序不同的缘故。


最后,写一个极端一点的例子,先后调用 doFinally、doAfterTerminate、doFinally、doAfterTerminate:

Observable.just("Hello World")
                .doOnNext(s -> System.out.println("doOnNext:"))
                .doFinally(() -> System.out.println("doFinally1:"))
                .doAfterTerminate(() -> System.out.println("doAfterTerminate1:"))
                .doFinally(() -> System.out.println("doFinally2:"))
                .doAfterTerminate(() -> System.out.println("doAfterTerminate2:"))
                .subscribe(
                        s -> System.out.println("onNext:" + s),
                        throwable -> System.out.println("onError:"),
                        () -> System.out.println("onComplete:"));


执行结果:

doOnNext:
onNext:Hello World
onComplete:
doAfterTerminate2:
doFinally2:
doAfterTerminate1:
doFinally1:


在 onComplete 调用之后,先打印了"doAfterTerminate2:",再打印"doFinally2:",然后打印"doAfterTerminate1:",最后打印"doFinally1:"。这正好符合刚才的分析。


总结



本文是对 doFinally 和 doAfterTerminate 两个操作符的总结。


相关文章
|
3月前
|
安全 Android开发
你是否了解 RxJava 的 Disposable ?
你是否了解 RxJava 的 Disposable ?
75 0
|
数据处理
RxJava2实现RxBus
RxJava2实现RxBus
143 0
|
安全 Android开发
详解 RxJava 的 Disposable
RxJava2 的 Disposable,可以在适当时机取消订阅、截断数据流,避免 Android 中的内存泄露。
982 0
|
Java Go Android开发
RxJava2
函数式编程是一种编程范式。我们常见的编程范式有命令式编程、函数式编程和逻辑式编程。我们常见的面向对象编程是一种命令式编程。命令式编程是面向计算机硬件的抽象,有变量、赋值语句、表达式和控制语句。而函数式编程是面向数学的抽象,将计算描述为一种表达式求值,函数可以在任何地方定义,并且可以对函数进行组合。响应式编程是一种面向数据流和变化传播的编程范式,数据更新是相关联的。把函数式编程里的一套思路和响应式编程合起来就是函数响应式编程。函数响应式编程可以极大地简化项目,特别是处理嵌套回调的异步事件、复杂的列表过滤和变换或者时间相关问题。在Android开发中使用函数响应式编程的主要有两大框架:
144 0
RxJava2
|
设计模式 存储 Java
XTask与RxJava的使用对比
XTask与RxJava的使用对比
135 0
XTask与RxJava的使用对比
|
负载均衡 算法 Java
RxJava 并行操作
RxJava 并行操作
377 0
RxJava 并行操作
|
Java API
RxJava 之 ParallelFlowable
RxJava 之 ParallelFlowable
218 0
RxJava 之 ParallelFlowable
|
Java
冷饭热炒——RxJava
已经超过一个月没有写文章了,原因无非就是工作太忙。最近终于恢复以前的节奏,任务开始正常了起来。忙里偷闲,写一写人们写烂了的RxJava。这篇文章主要分析RxJava事件的产生以及变化的原理,Ok,let's go! 0. 前言 本次源码分析使用的是RxJava2,版本2.1.14。
1034 0
|
Java 数据库 Android开发
06.RxJava初探
基本使用 在build.gradle中加入配置,注意,rxJava和rxAndroid版本一定要相互兼容,不然可能会报错More than one file was found with OS independent path 'META-INF/rxjava.
1154 0
RxJava2学习笔记(3)
接上回继续,今天来学习下zip(打包)操作 一、zip操作 @Test public void zipTest() { Observable.zip(Observable.
1076 0