RxJava2.0的使用详解

2018-06-25  本文已影响38人  CarlosLynn

RxJava2.0的使用详解

1,初识RxJava

RxJava就是一种用Java语言实现的响应式编程，来创建基于事件的异步程序
RxJava是一个基于事件订阅的异步执行的一个类库,目前比较火的一些技术框架！

1.1使用前所添加的依赖(build.gradle):

implementation 'io.reactivex.rxjava2:rxjava:2.1.6'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

1.2作用:

RxJava的目的就是异步。
RxJava的特点就是可以非常简便的实现异步调用，可以在逻辑复杂的代码逻辑中以比较轻易的方式实现异步调用。随着逻辑的复杂，需求的更改，代码可依然能保持极强的阅读性

1.3概念:

RxJava是利用观察者模式来实现一些列的操作，所以对于观察者模式中的观察者，被观察者，以及订阅、事件需要有一个了解.
Observable：在观察者模式中称为“被观察者”；
Observer：观察者模式中的“观察者”，可接收Observable发送的数据；
subscribe：订阅，观察者与被观察者，通过Observable的subscribe()方法进行订阅；
Subscriber：也是一种观察者，在2.0中 它与Observer没什么实质的区别，不同的是 Subscriber要与Flowable(也是一种被观察者)联合使用，该部分 内容是2.0新增的，后续文章再介绍。Obsesrver用于订阅Observable，而Subscriber用于订阅Flowable.

1.4观察者模式的理解:

A 对象（观察者）对 B 对象（被观察者）的某种变化高度敏感,需要在 B 变化的一瞬间做出反应.
在程序的观察者模式，观察者不需要时刻盯着被观察者,而是采用注册或者称为订阅的方式，告诉被观察者：我需要你的某某状态，你要在它变化的时候通知我!
RxJava 有四个基本概念：
Observable (被观察者)、
Observer (观察者)、
subscribe (订阅)
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在完成某些操作，获得一些结果后，回调触发事件，即 发出事件来通知 Observer。
关于回调，如果理解则可以跳过这一段，如果不理解，在RxJava中可以简单的理解为：为了方便Observable和Observer交互，在Observable中，将 Observer对象传入，在完成某些操作后调用Observer对象的方法，此时将触发Observer中具体实现的对应方法。
注意：Observer是个接口，Observable是个类。

RxJava中定义的事件方法：

onNext(),普通事件,按照队列依次进行处理.
onComplete(),事件队列完结时调用该方法
onError(),事件处理过程中出现异常时，onError()触发，同时队列终止,不再有事件发出.
onSubscribe(),RxJava 2.0 中新增的，传递参数为Disposable,可用于切断接收事件
让Observable (被观察者)开启子线程执行耗操作，完成耗时操作后，触发回调，通知Observer (观察者)进行主线程UI更新

2,简单使用步骤:

步骤:
创建数据发射源,上游Observable
创建数据接收处,下游Observer
数据源关联接收处,上游衔接下游!

3,Observable

数据发射源,可观察的,被观察的,

Observable有两种形式启动形式:

1热启动Observable任何时候都会发送消息，即使没有任何观察者监听它。
2冷启动Observable只有在至少有一个订阅者的时候才会发送消息

Observable的几种创建方式:

部分方法介绍:

表示下游不关心任何事件,你上游尽管发你的数据

Disposable subscribe()

表示下游只关心onNext事件,其他不管

Disposable subscribe(Consumer<? super T> onNext)

表示下游只关心onNext事件,onError事件

Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)

表示只关心onNext事件,onError事件,onComplete事件

Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete)

表示处理所有事件

subscribe(Observer<? super T> observer)

ObservableEmitter

Emitter是发射器的意思,就是用来发出事件的，它可以发出三种类型的事件

Disposable

一次性,它理解成两根管道之间的一个机关, 当调用它的dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件.
在RxJava中,用它来切断Observer(观察者)与Observable(被观察者)之间的连接，当调用它的dispose()方法时, 它就会将Observer(观察者)与Observable(被观察者)之间的连接切断, 从而导致Observer(观察者)收不到事件。
注意:
调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件
我们让上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 切断水管, 看看运行结果

Disposable的对象通过观察者获得，具体分为两种方式
1,Observer接口

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                //此方法接收到Disposable的实例!
            }

            @Override
            public void onNext(String s) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };

通过创建Observer(观察者)接口,重写onSubscribe方法,当订阅后，建立与Observable(被观察者)的联系后，在onSubscribe(Disposable d)方法中便可以获得Disposable对象。
2.Consumer等其他函数式接口

    Disposable disposable = Observable.just("你好").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {

            }
        });

当调用Observable的subscribe()方法后直接返回一个Disposable 对象

6,线程控制——Scheduler

在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。

Schedulers.immediate():
直接在当前线程运行，相当于不指定线程。这是默认的Scheduler。

Schedulers.newThread():
总是启用新线程，并在新线程执行操作。

Schedulers.io(): I/O
操作（读写文件、读写数据库、网络信息交互等）所使用的Scheduler。行为模式和newThread()差不多，区别在于io()的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下io()比newThread()更有效率。不要把计算工作放在io()中，可以避免创建不必要的线程。

Schedulers.computation():
计算所使用的Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个Scheduler使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在computation()中，否则 I/O 操作的等待时间会浪费 CPU。

AndroidSchedulers.mainThread()，
Android专用线程，指定操作在主线程运行。

如何切换线程呢？RxJava中提供了两个方法：
subscribeOn() 和 observeOn() ，
两者的不同点在于：

subscribeOn():
指定subscribe()订阅所发生的线程，或者叫做事件产生的线程。

observeOn():
指定Observer所运行在的线程，即onNext()执行的线程。或者叫做事件消费的线程。

7,以Consumer为例，我们可以实现简便式的观察者模式

Observable.just("hello").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s);
            }
        });

其中Consumer中的accept()方法接收一个来自Observable的单个值。Consumer就是一个观察者。其他函数式接口可以类似应用

8,RxJava中的操作符

01,操作符就是用于在Observable和最终的Observer之间，通过转换Observable为其他观察者对象的过程，修改发出的事件，
最终将最简洁的数据传递给Observer对象.
每次调用一次操作符，就进行一次被观察者对象的改变，同时将需要传递的数据进行转变，最终Observer对象获得想要的数据。
以网络加载为例，我们通过Observable开启子线程，进行一些网络请求获取数据的操作，获得到网络数据后，然后通过操作符进行转换，获得我们想要的形式的数据，然后传递给Observer对象

02,比较常用的操作符:

Observable<Integer> observable = Observable
.just("hello")
.map(new Function<String, Integer>() {
    @Override
    public Integer apply(String s) throws Exception {
        return s.length();
    }
});

Observable.just(list)
       .flatMap(new Function<List<String>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(List<String> strings) throws Exception {
                return Observable.fromIterable(strings);
            }
        });

Observable
        .just(list)
        .flatMap(new Function<List<String>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(List<String> strings) throws Exception {
                return Observable.fromIterable(strings);
            }
        }).filter(new Predicate<Object>() {
            @Override
            public boolean test(Object s) throws Exception {
                String newStr = (String) s;
                if (newStr.charAt(5) - '0' > 5) {
                    return true;
                }
                return false;
            }
        }).subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object o) throws Exception {
                System.out.println((String)o);
            }
        });

        Observable.just(new ArrayList<String>(){
            {
                for (int i = 0; i < 8; i++) {
                    add("data"+i);
                }
            }
        }).flatMap(new Function<List<String>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(List<String> strings) throws Exception {
                return Observable.fromIterable(strings);
            }
        }).take(5).subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object s) throws Exception {
                DemonstrateUtil.showLogResult(s.toString());
            }
        });

 Observable.just(new ArrayList<String>(){
            {
                for (int i = 0; i < 6; i++) {
                    add("data"+i);
                }
            }
        }).flatMap(new Function<List<String>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(List<String> strings) throws Exception {
                return Observable.fromIterable(strings);
            }
        }).take(5).doOnNext(new Consumer<Object>() {
            @Override
            public void accept(Object o) throws Exception {
                DemonstrateUtil.showLogResult("额外的准备工作!");
            }
        }).subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object s) throws Exception {
                DemonstrateUtil.showLogResult(s.toString());
            }
        });

8,Flowable的理解

Flowable是一个被观察者，与Subscriber(观察者)配合使用，解决Backpressure问题
Backpressure(背压)。所谓背压，即生产者的速度大于消费者的速度带来的问题。

什么情况下才会产生Backpressure问题？

现象演示说明:

被观察者是事件的生产者,观察者是事件的消费者.假如生产者无限生成事件,而消费者以很缓慢的节奏来消费事件,会造成事件无限堆积,形成背压,最后造成OOM!
Flowable悠然而生，专门用来处理这类问题。
Flowable是为了应对Backpressure而产生的。Flowable是一个被观察者，
与Subscriber(观察者)配合使用，解决Backpressure问题。
注意：处理Backpressure的策略仅仅是处理Subscriber接收事件的方式，并不影响Flowable发送事件的方法。
即使采用了处理Backpressure的策略，Flowable原来以什么样的速度产生事件，现在还是什么样的速度不会变化，主要处理的是Subscriber接收事件的方式。

处理Backpressure问题的策略,或者来解决Backpressure问题

BackpressureStrategy.ERROR
如果缓存池溢出,就会立刻抛出MissingBackpressureException异常
request()用来向生产者申请可以消费的事件数量,这样我们便可以根据本身的消费能力进行消费事件.
虽然并不限制向request()方法中传入任意数字，但是如果消费者并没有这么多的消费能力，依旧会造成资源浪费，最后产生OOM
at java.lang.OutOfMemoryError.<init>(OutOfMemoryError.java:33)
在异步调用时，RxJava中有个缓存池，用来缓存消费者处理不了暂时缓存下来的数据，缓存池的默认大小为128，即只能缓存128个事件。
无论request()中传入的数字比128大或小，缓存池中在刚开始都会存入128个事件。
当然如果本身并没有这么多事件需要发送，则不会存128个事件。
应用举例:

BackpressureStrategy.BUFFER
是把RxJava中默认的只能存128个事件的缓存池换成一个大的缓存池,支持存更多的数据.
消费者通过request()即使传入一个很大的数字，生产者也会生产事件,并将处理不了的事件缓存.
注意:
这种方式任然比较消耗内存，除非是我们比较了解消费者的消费能力，能够把握具体情况，不会产生OOM。
BUFFER要慎用

    Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                while (true){
                    e.onNext(1);
                }
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Thread.sleep(2000);
                System.out.println(integer);
            }
        });

被观察者是事件的生产者，观察者是事件的消费者。上述例子中可以看出生产者无限生成事件，而消费者每2秒才能消费一个事件，这会造成事件无限堆积，最后造成OOM。
Flowable就是由此产生，专门用来处理这类问题

image.png image.png

参考资料:

Github上RxJava的项目地址:
https://github.com/ReactiveX/RxJava
技术文档Api:
http://reactivex.io/RxJava/javadoc/

RxAndroid，用于 Android 开发:
https://github.com/ReactiveX/RxAndroid
简书博客推荐:
http://www.jianshu.com/p/ba61c047c230

RxJavaDemo

上一篇 下一篇

猜你喜欢

热点阅读