RxJava 之 响应式编程

时间:2022-08-27 14:07:19

什么是RxJava

  • RxJava 就是异步
  • RxJava 的异步实现,是通过一种扩展的观察者模式来实现的。
  • 一个响应式编程框架

响应式编程

在说响应式编程的概念之前,我们先从这篇文章《用工厂流水线的方式来理解 RxJava 的概念》理解一下RxJava的意义。这里面说到的流水线的例子很确切,对于RxJava的响应式编程的思想是一个很好地类比。

从这个例子中,我们可以抽象出如下的流程

流水线——->操作工1—–>操作工2—–>…..操作工n–>产品经理

这也十分切合响应式编程的流程

Observable -> Operator 1 -> Operator 2 -> Operator 3 -> Subscriber

  • 如文中所讲,产品经理很重要,推动了整个事件的方法,这一点也是RxJava 区别于传统的观察者模式的地方,没有订阅者(subscriber)时,Observeable 是不会发送事件的。

  • 作为操作符,就像流水线上的工人,既要各司其职的完成各种工作,同时也要保证事件的处理永远处于流水线上,这也是RxJava响应式编程的关键点,“流”的概念。

  • 在这里,产品经理(Subscriber )也会完成一些重要的工作,但它希望流水线上的每个操作工(RxJava的操作符)也能完成一些重要的事情,能最大程度的确保产品经理已经完成的工作不需要再做修改。

下面结合一个简单的例子说明一下,RxJava中响应式编程的优点。

RxJava 事件处理

假设,现有如下需求,输出下面数组中每个数与 10 的余数

List<Integer> numbers = new ArrayList<>();
        for (int i = 5; i >= 0; i--) { numbers.add(i); }

用RxJava实现如下:

       Observable.from(numbers)
                .map(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        return 10 % integer;
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.e(MainActivity.class.getSimpleName(), "onCompleted");
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.e(MainActivity.class.getSimpleName(), "onError---->" + e.getMessage());
                    }
                    @Override
                    public void onNext(Integer integer) {
                        Log.e(MainActivity.class.getSimpleName(), "onNext---- >Remainder is " + integer);
                    }
                });

输出:

RxJava 之 响应式编程

可以看到,这里出错了,数组中有一项为0 ,导致了除数为0,发生了异常,但是程序并没有崩溃,而是抛出了异常信息。同时onError方法执行了,而onComplete()方法没有执行。

我们修复一下这个错误(这里假设我们无法修改初始数据,也就是这里的numbers数组)

Observable.from(numbers)
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer!=0;
                    }
                })
                .map(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        return 10 % integer;
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.e(MainActivity.class.getSimpleName(), "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(MainActivity.class.getSimpleName(), "onError---->" + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(MainActivity.class.getSimpleName(), "onNext---- >Remainder is " + integer);
                    }
                });

输出如下:

RxJava 之 响应式编程

我们使用RxJava的操作符filter 过滤掉了0,这样就不会有错误,最终执行了onCompleted方法。

从上面的例子中,我们得出如下结论:

  1. 在一次完整的事件中,onNext可能会执行0-n次,而onError() 和 onCompleted()中必有一个方法会执行。
  2. 有异常发生,onError就会调用
  3. 订阅者如果完成了所有事件的处理,onCompleted就会执行,使得我们可以得知整个事件结束了
  4. 操作符不负责处理异常,而是由订阅者处理。

这里想一下,按照以往的方式,发生错误时,通常是在每个回调中处理错误。这不仅导致了重复的代码,并且意味着每个回调都必须知道如何处理错误,你的回调代码将和调用者紧耦合在一起。

使用RxJava,Observable对象根本不需要知道如何处理错误!操作符也不需要处理错误状态-一旦发生错误,就会跳过当前和后续的操作符。所有的错误处理都交给订阅者来做。

但是,这种机制也不是完美,有时候我们可能需要对不同的错误做出不同的处理,RxJava的机制反而不适合了。

这里只是对响应式编程思想的简单了解!