RxJava 知识梳理(1) RxJava 基本思想

时间  2019-11-21
标签 rxjava 知识 梳理 基本 思想 栏目 Java 繁體版
原文   原文链接

1、基础概述

RxJava的关键是异步，即便随着程序的逻辑变得复杂，它依然可以保持简洁。bash

2、API介绍和原理剖析

观察者模式面向的需求是：A对象（观察者）对B对象（被观察者）的某种变化高度敏感，须要在B变化的一瞬间作出反应，观察者采用注册Register或者订阅Subscribe的方式，告诉观察者，我须要你的某某状态，并在它变化的时候通知我，在RxJava当中，Observable是被观察者，Observer就是观察者。异步

RxJava有四个基本概念：ide

  • Observable：被观察者。
  • Observer：观察者。
  • Subscribe：订阅。
  • Event：事件。

Observable和Observer经过subscribe方法实现订阅关系，Observable能够在须要的时候发出事件来通知Observer。函数

RxJava有如下三种事件：ui

  • onNext：普通事件。
  • onCompleted：RxJava不只把每一个事件单独处理，还会把它们看做一个队列，当不会再有新的onNext事件发出时，须要触发onCompleted事件做为标志。
  • onError：onCompleted和有且仅有一个，而且是事件序列中的最后一个。

3、基本实现

RxJava的基本实现有如下三点： 1）建立观察者 - Observerspa

Observer<String> observer = new Observer<String>() {

    @Override
    public void onCompleted() {
        Log.d(TAG, "onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "onError");
    }

    @Override
    public void onNext(String s) {
        Log.d(TAG, "onNext");
    }
};
复制代码

除了Observer接口以外，RxJava还内置了一个实现了Observer的抽象类：Subscriber，它对Observer接口进行了一些扩展，实质上在RxJava的subscribe过程当中，Observer也老是被转换成为一个Subscriber再使用，他们的区别在与：线程

  • onStart：这是新增的方法，它会在subscribe刚开始，而事件还未发送以前被调用，它老是在subscribe所发生的线程被调用。
  • unsubscribe：这是它实现的另外一个接口Subscription的方法，用于取消订阅，在这个方法被调用后，Subscriber将再也不接收事件，通常在调用这个方法前，可使用isUnsubscribed判断一下状态，Observable在订阅以后会持有Subscriber的引用，所以不释放会有内存泄漏的危险。

2）建立被观察者 - Observable RxJava用create方法来建立一个observable，3d

rx.Observable observable = rx.Observable.create(new rx.Observable.OnSubscribe<String>() {

    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello World!");
        subscriber.onCompleted();
    }
});
复制代码

这里传入了一个Observable.OnSubscribe<T>对象做为参数，它会被存储在返回的Observable对象当中，它的做用至关于一个计划表，当Observable被订阅的时候，OnSubscribe的call方法会自动被调用，事件序列被依次触发。 create是RxJava最基本的创造事件序列的方法，基于这个方法，还提供了一些快捷方法来建立事件队列：代理

  • just(T...)
Observable observable = Observable.just("Hello", "Hi", "Aloha");
复制代码
  • from(T[]) / from(Iterable<? extends T>)
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
复制代码

3）订阅 - subscribecode

observable.subscribe(observer);
observable.subscribe(subscriber);
复制代码

其内部核心的代码相似于：

public Subscription subscribe(Subscriber subscriber) {
    //准备方法。
    subscriber.onStart();
    //事件发送的逻辑开始执行，这个onSubscribe就是建立Observable时新建的OnSubscribe对象。
    onSubscribe.call(subscriber);
    //把传入的Subscriber转换为Subscription并返回，方便unsubscribe。
    return subscriber;
}
复制代码

Observable.subscribe方法除了支持传入Observer和Subscriber，还支持传入Action0、Action1这样不完整定义的回调，RxJava会自动根据定义建立出Subscriber。

4、线程控制

在不指定线程的状况下，RxJava遵循这样的原则，在哪一个线程调用subscribe，就在哪一个线程产生事件，在哪一个线程产生事件，就在哪一个线程消费事件，若是须要消费线程，那么就须要用到Scheduler， RxJava内置了几个Scheduler：

  • Schedulers.immediate：直接在当前线程运行。
  • Schedulers.newThread：老是启用新线程，并在线程执行操做。
  • Schedulers.io：其内部实现是一个无数量上限的的线程池，能够重用空闲的线程，不要把计算工做放在io，能够避免建立没必要要的线程。
  • Schedulers.computation：使用固定的线程池，大小为CPU核数。
  • AndroidSchedulers.mainThread：指定的操做将在Android主线程中运行。

对线程控制有如下两个方法：

  • subscribeOn：指定subscribe发生的线程，即Observable.OnSubscribe被激活时所处的线程，也就是call方法执行时所处的线程。
  • observeOn：指定Subscriber所运行在的线程。

observeOn指定的是Subscriber的线程，而这个Subscriber并不必定是subscribe()参数中的Subscriber，而是observeOn执行时的当前Observable所对应的Subscriber，即它的直接下级Subscriber，也就是它以后的操做所在的线程，所以，若是有屡次切换线程的要求，只要在每一个想要切换线程的位置调用依次observeOn便可。 和observeOn不一样，subscribeOn只能调用一次，下面咱们来分析一下它的内部实现，首先是subscribeOn的原理： subscribeOn和ObserveOn都作了线程切换的工做：

  • subscribeOn的线程切换发生在OnSubscribe中，即在它通知上一级的OnSubscribe时，这时事件尚未发送，所以subscribeOn的线程控制能够从事件发出的开端形成影响。

Paste_Image.png

  • observeOn的线程切换则发生在它内建的Subscriber中，即发生在它即将给下一级Subscriber发送事件时，所以控制的是它后面的线程。

Paste_Image.png

5、变换

变换，就是将事件序列中的对象或整个序列进行加工处理，转换不一样的事件或者序列。

5.1 map()

经过FuncX，把参数中的Integer转换成为String，是最经常使用的变换，这个变换是发生在subscribeOn所指定的线程当中的。

Subscriber<String> subscriber = new Subscriber<String>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(String s) {
        long nextId = Thread.currentThread().getId();
        Log.d(TAG, "onNext:" + s + ", threadId=" + nextId);
    }
};
Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        long callId = Thread.currentThread().getId();
        subscriber.onNext(5);
        subscriber.onCompleted();
    }
});
observable.map(new Func1<Integer, String>() {

    @Override
    public String call(Integer integer) {
        long mapId = Thread.currentThread().getId();
        return "My Number is:" + integer;
    }
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);
复制代码

其示意图相似于：

Paste_Image.png

5.2 flatMap

它和map有一个共同点，就是把传入的参数转化以后返回另外一个对象，可是和map不一样的是，flatMap返回的是一个Observable对象，并且它并不直接把这个对象传给Subscriber，而是经过这个新建的Observable来发送事件，其整个的调用过程：

  • 使用传入的事件对象建立一个Observable。
  • 激活这个Observable，经过它来发送事件。
  • 每个建立出来的Observable发送的事件，被汇入同一个Observable，它复杂将这些事件同一交给Subscriber的回调方法。
Subscriber<String> subscriber = new Subscriber<String>() {

    @Override
    public void onCompleted() {}

    @Override
    public void onError(Throwable e) {}

    @Override
    public void onNext(String s) {
        Log.d(TAG, "onNext, s=" + s);
    }
};
Observable<List<String>> observable = Observable.create(new Observable.OnSubscribe<List<String>>() {

    @Override
    public void call(Subscriber<? super List<String>> subscriber) {
        List<String> list = new ArrayList<>();
        list.add("First");
        list.add("Second");
        list.add("Third");
        subscriber.onNext(list);
    }
});
observable.flatMap(new Func1<List<String>, Observable<String>>() {
    @Override
    public Observable<String> call(List<String> strings) {
        return Observable.from(strings);
    }
}).subscribe(subscriber);
复制代码

其示意图：

Paste_Image.png

6、变换的原理

变换的实质是针对事件序列的处理和再发送，在RxJava的内部，它们是基于同一个基础的变换方法lift(operator)

//生成了一个新的Observable并返回。
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
    //构造新的Observable时，同时新建了一个OnSubscribe对象。
    return Observable.create(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber subscriber) {
            Subscriber newSubscriber = operator.call(subscriber);
            newSubscriber.onStart();
            //原始的onSubscribe。
            onSubscribe.call(newSubscriber);
        }
    });
}
复制代码

示意图：

Paste_Image.png

  • lift建立了一个Observable后，加上以前的原始Observable，有两个Observable。
  • 新的Observable里的OnSubscribe加上原始的，共有两个OnSubscribe。
  • 当用户经过调用lift/map建立的Observable对象的subscribe方法时，因而它触发了上面的call方法中的内容。
  • 在这个新的OnSubscribe的call方法中，传入了目标的Subscriber，同时其外部类中还持有了原始的OnSubscribe。咱们先经过operator.call(oldSubscriber)方法，生成了新的Subscriber（new Subscriber），而后利用这个新的Subscriber向原始的Observable进行订阅。

下面咱们之前面map实现的例子来分析一下源码，上面的例子经过map操做符把Integer类型的Observable和String类型的Subscriber生成了订阅关系。

  • map方法，它经过lift方法返回了一个String类型的Observable。
//其中T=Integer，R=String。
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
        return lift(new OperatorMap<T, R>(func));
}
复制代码
  • 下面看下OperatorMap这个对象，这个对象实现了operator<R,T>接口，而这个接口继承于Func1<Subscriber<? super R>, Subscriber<? super T>>，在它实现的call方法中传入了String类型的Subscriber（目标Subscriber），并返回了Integer类型的Subscriber（代理Subscriber），当它的方法被回调时，会调用目标Subscriber的对应方法，其中在调用onNext时，就用上了外部传入的Func1函数：
@Override
    public Subscriber<? super T> call(final Subscriber<? super R> o) {
        return new Subscriber<T>(o) {

            @Override
            public void onCompleted() {
                o.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                o.onError(e);
            }

            @Override
            public void onNext(T t) {
                try {
                    o.onNext(transformer.call(t));
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    onError(OnErrorThrowable.addValueAsLastCause(e, t));
                }
            }

        };
    }
复制代码
  • 接着再回过头来看lift方法：
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return new Observable<R>(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> o) {
                try {
                    //返回一个Integer类型的Subscriber。
                    Subscriber<? super T> st = hook.onLift(operator).call(o);
                    try {
                        st.onStart();
                        //关键方法：Integer类型的OnSubscribe调用对应的Subscribe，这个call方法里面写了咱们的逻辑，当它调用onNext(Integer integer)时，实际上调用的是onNext(String str)。
                        onSubscribe.call(st);
                    } catch (Throwable e) {
                        if (e instanceof OnErrorNotImplementedException) {
                            throw (OnErrorNotImplementedException) e;
                        }
                        st.onError(e);
                    }
                } catch (Throwable e) {
                    if (e instanceof OnErrorNotImplementedException) {
                        throw (OnErrorNotImplementedException) e;
                    }
                    o.onError(e);
                }
            }
        });
    }
复制代码
  • 最后就是调用subscribe方法。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程