RxJava学习笔记

作者: shuixingge | 来源:发表于2016-06-04 22:19 被阅读177次

本文仅为学习笔记；不是原创文章；

一：基本介绍

概念： 是一个实现异步操作的库
优势： 使逻辑变得简洁，而不是代码量得增减，当逻辑变得复杂时，RxJava能保证从上到下仍然是链式调用，没有任何嵌套。

二：原理

概念： 扩展的观察者模式
经典的观察者模式：点击监听器onClickListener；其中设置了onClickListener的View是被观察者；onClickListener是观察者，两者通过setOnClickListener()来达成订阅关系；订阅之后用户点击按钮的瞬间，Android Framework 就会将点击事件发送给已经注册的 OnClickListener。采取这样被动的观察方式，既省去了反复检索状态的资源消耗，也能够得到最高的反馈速度。

经典的观察者模式
RxJava 的观察者模式： 使用的就是通用形式的观察者模式
RxJava有四个基本概念：Observable(被观察者)，Observer(观察者)，subscribe(订阅)，事件。Observable和Observer通过subscribe达成订阅关系，从而 Observable可以在需要的时候发出事件来通知 Observer。
RxJava定义了三个事件回调方法：
(1) onNext()：相当于onClick(),每个单独的事件处理回调方法
(2) onCompleted()： 事件队列完结。RxJava不仅把每个事件单独处理，还会把它看成一个事件队列。当不再有新的onNext()发出时，需要触发onCompleted()作为结束标志。
(3) onError()： 事件队列异常，在事件处理过程中出异常时，onError()
会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted()和 onError()有且只有一个，并且是事件序列中的最后一个。

RxJava 的观察者模式

三：基本实现

3.1 创建Observer

Observer 即观察者，它决定事件触发的时候将有怎样的行为。可以通过创建Observer对象和创建Subscriber对象来实现观察者。

3.1.1 创建Observer对象：

 Observer<String> observer=new Observer<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {

            }
        };

3.1.2 创建Subscriber对象：

 Subscriber<String> subscriber=new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {

            }
        };

Subscriber实现了 Observer，并对 Observer 接口进行了一些扩展，实质上，在 RxJava 的 subscribe 过程中，Observer也总是会先被转换成一个 Subscriber再使用。

public abstract class Subscriber<T> implements Observer<T>, Subscription

public final Subscription subscribe(final Observer<? super T> observer) {
          //转换成Subscriber
        if (observer instanceof Subscriber) {
            return subscribe((Subscriber<? super T>)observer);
        }
        return subscribe(new Subscriber<T>() {

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

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

            @Override
            public void onNext(T t) {
                observer.onNext(t);
            }

        });
    }

3.1.3 Subscriber和Observer的区别
onStart()： 这个是Subscriber新增的方法，用于subscribe刚开始,事件还没发送之前，可以用于做一些准备工作。例如事件清零。onStart方法总是在subscribe的线程发生。
unSubscrible()： 这个是Subscribe实现的Subscription接口中的方法，用于取消订阅，因为subscribe后， Observable会持有一个Subscriber的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop()等方法中）调用unsubscribe()来解除引用关系，以避免内存泄露的发生。

3.2 创建Observable

3.2.1 使用 create方法创建： observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。RxJava使用create方法来创建一个Observable,其中OnSubcribe对象作为参数传入。当Observable被订阅时，OnSubcribe对象的call()方法会被自动调用，事件会按照设定依次触发。

 Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});

RxJava 还提供了一些方法用来快捷创建事件队列
3.2.2 使用just(T...) 创建： 将传入的参数依次发送出来。

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

3.2.3 使用from(T[]) /from(Iterable<? extends T>)创建：将传入的数组或者Iterable拆分成具体的对象后，依次发送出来。

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

3.3 Subcribe(订阅)

创建了 Observable 和 Observer之后，再用 subscribe() 方法将它们联结起来。

observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);

Observable.subscribe(Subscriber) 的内部实现是这样的（逻辑实现）：

public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    onSubscribe.call(subscriber);
    return subscriber;
}

主要做了三件事：
(1)调用 subscriber.onStart();
(2)调用onSubscribe.call(subscriber);
(3)将传入的 Subscriber 作为 Subscription返回。

订阅事件

四：RxJava不完整定义的回调(Action0,Action1)。

4.1 ActionX

4.1.1 Action0： 是一个接口，只有一个call()方法，并且该方法无参无返回值。

public interface Action0 extends Action {
    void call();
}

4.1.2 Action1： 是一个接口，只有一个call()方法，并且该方法有一个参数返回值。

public interface Action1<T> extends Action {
    void call(T t);
}

RxJava还提供了Action3等ActionX，同样只有一个call()方法，只是参数个数不一样。

4.2 ActionX的使用

因为onCompleted()无参无返回值，所以Action0可以将onCompleted()打包起来，将自己的一个参数传入subscribe()来实现不完整回调;
同样因为onError()和onCompleted()没有返回值只有一个参数，所以Action1可以将onError()和onCompleted()打包起来，传入subscribe()来实现不完整回调；类似于闭包的原理

Action1<String> onNextAction = new Action1<String>() {
    // onNext()
    @Override
    public void call(String s) {
        Log.d(tag, s);
    }
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    // onError()
    @Override
    public void call(Throwable throwable) {
        // Error handling
    }
};
Action0 onCompletedAction = new Action0() {
    // onCompleted()
    @Override
    public void call() {
        Log.d(tag, "completed");
    }
};

// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

五：线程控制(Scheduler)

5.1 Scheduler API

RxJava 遵循的是线程不变的原则：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。
切换线程需要通过Scheduler，RxJava内置的Schedulers。
Schedulers.immediate()：直接在当前线程运行。
Schedulers.newThread()： 启动新线程，并在新线程执行操作。
Schedulers.io()： I/O操作使用的Scheduler。
Schedulers.computation()：计算所使用的Scheduler，CPU密集型计算，即不会被I/O操作限制性能的操作，不要把I/O操作放在这个computation()中，否则这个I/O操作的等待时间会浪费CPU。
AndroidSchedulers.mainThread()：使操作执行在Android的主线程。
subscribeOn()： 指定subscribe发生的线程，即事件生产线程。
observeOn()： 指定Subscriber所运行的线程，即事件消费线程。

5.2 变换(Map)

5.2.1 概念： 就是将事件序列中的对象或者整个事件序列进行加工处理，转换成不同的事件或者事件序列。
5.2.2 FunX和ActionX： FunX和ActionX很类似，都是RxJava中的接口，都有一个call方法，X的大小表示参数的多少，不同的是FunX的call()方法是有返回值的。

public interface Func0<R> extends Function, Callable<R> {
    @Override
    R call();
}

public interface Func1<T, R> extends Function {
    R call(T t);
}

5.2.3 一个Map的例子： map()方法可以事件对象的类型转换成另一种类型对象后返回；一对一的转化；事件对象的直接变换

Observable.just("images/logo.png") // 输入类型 String
    .map(new Func1<String, Bitmap>() {
        @Override
        public Bitmap call(String filePath) { // 参数类型 String
            return getBitmapFromPath(filePath); // 返回类型 Bitmap
        }
    })
    .subscribe(new Action1<Bitmap>() {
        @Override
        public void call(Bitmap bitmap) { // 参数类型 Bitmap
            showBitmap(bitmap);
        }
    });

5.2.4 flatMap的例子：
flatMap和map的一个共同点就是把传入的参数转化之后返回另一个对象。和map不同的是flatMap返回的是一个Observable对象，并且这个Observable对象并不是直接发送到Subcriber的回调方法。
flatMap的原理：
(1)使用传入的事件对象创建一个Observable对象。

由于student是一个数组，所以会创建多个Observable<Course>
return Observable.from(student.getCourses())

(2)并不直接发送这个Obserable对象，而是将它激活，于是它开始发送事件。

(3) 每一个创建出来的Obserable对象所发送的事件都被汇入同一个Obserable对象中，而由这个Obserable对象将事件统一的发给Subscriber的回调方法。

 Student[] students = ...;
 Subscriber<Course> subscriber = new Subscriber<Course>() {
    @Override
    public void onNext(Course course) {
        Log.d(tag, course.getName());
    }
    ...
};
Observable.from(students)
    .flatMap(new Func1<Student, Observable<Course>>() {
        @Override
        public Observable<Course> call(Student student) {
            return Observable.from(student.getCourses());
        }
    })
    .subscribe(subscriber);

5.2.5 Map和flatMap变换的原理：lift()
lift()核心代码(逻辑代码)：
创建了一个新的Observable对象。

 public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
    return Observable.create(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber subscriber) {
            Subscriber newSubscriber = operator.call(subscriber);
            newSubscriber.onStart();
            onSubscribe.call(newSubscriber);
        }
    });
}

对照之前的Observable.subscribe()可以发现两者几乎看起来一样，但是需要注意的是onSubscribe对象不一样
Observable.subscribe()：Observable.subscribe()中的onSubscribe指的就是Observable中的onSubscribe对象。
lift()：
(1) lift()会创建一个新的Observable，加上原始的Observable就会有两个Observable。
(2)两个Observable拥有两个onSubscribe对象。
(3)当用户调用经过lift()变换之后的Observable.subscribe(),触发的是新的onSubscribe的call()方法，即在 lift()中生成的那个。
(4) 而这个新的onSubscribe的call()方法中的 onSubscribe指的是原始的Observable 中的原始 OnSubscribe，在新的onSubscribe的call()方法里面，利用operator.call方法生成一个新的Subscriber，然后使用这个新的newSubscriber向原始的 Observable进行订阅。

public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    onSubscribe.call(subscriber);
    return subscriber;
}

lift原理
5.2.6 compose()
lift() 是针对事件项和事件序列的，而 compose()是针对 Observable自身进行变换。

六：线程控制：Scheduler (二)

observeOn()： 可以多次调用；指定的是它之后的操作所在的线程。observeOn()指定的是Subscriber的线程，而这个Subscriber并不一定是subscribe()中的Subscriber，而是observeOn()执行时，当前Obserable所对应的Subscriber。
subscribeOn()： subscribeOn()的位置放在哪里都可以，但它是只能调用一次的。
subscribeOn()和observeOn()的原理
subscribeOn 的线程切换发生在OnSubcribe中，即在它通知上一级 OnSubscribe 时，这时事件还没有开始发送，因此subscribeOn()的线程控制可以从事件发出的开端就造成影响；
observeOn() 的线程切换则发生在它内建的 Subscriber中，即发生在它即将给下一级 Subscriber 发送事件时，因此 observeOn()控制的是它后面的线程。
doOnSubscribe()
默认情况下doOnSubscribe()执行在subscribe()线程；但是如果doOnSubscribe()后面有subscribeOn()的话，它将执行在离它最近的 subscribeOn()所指定的线程。
doOnSubscribe()主要是在事件流程之前进行一些初始化工作，和Subscriber的 onStart()方法左右类似1；不同的是onStart()方法不能指定线程，它执行在 subscribe()被调用时的线程，这会对onStart()中代码有要求。

RxJava学习笔记

本文仅为学习笔记；不是原创文章；

一：基本介绍

二：原理

三：基本实现

3.1 创建Observer

3.2 创建Observable

3.3 Subcribe(订阅)

四：RxJava不完整定义的回调(Action0,Action1)。

4.1 ActionX

4.2 ActionX的使用

五：线程控制(Scheduler)

5.1 Scheduler API

5.2 变换(Map)

六：线程控制：Scheduler (二)

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

知识 | 解析