0
点赞
收藏
分享

微信扫一扫

RxJava原理学习


前言

本文为个人学习RxJava框架时所作,参考了两篇大佬的文章,在文末给出了链接。我只是在这里做一个两方面的整合,方便个人学习。

总结

RxJava是一个响应式编程框架,可以非常方便的实现线程调度以及数据转换,规范代码,并且其用到了观察者模式和装饰者模式。与Retrofit框架配合较多。

1.RxJava是什么?

响应式编程框架,响应式编程是一种通过异步和数据流来构建事务关系的编程模型。能够由事件来驱动事务
RxJava利用响应式编程的优点,可以实现
①可以通过一系列链式调用,规范代码
②能够通过​​​subscribeOn​​​和​​observeOn​​​两个方法实现线程切换,通过​​Map,FlatMap​​​等操作符实现中间事件的转换。
③与Retrofit框架配合使用较多

2.RxJava的线程调度

线程调度只有​​subscribeOn()​​​和​​observeOn()​​​两个方法。
①​​​subscribeOn()​​​它指示​​Observable​​​在一个指定的调度器上创建(只作用于被观察者创建阶段)。只能指定一次,如果指定多次则以第一次为准
②​​​observeOn()​​指定在事件传递(加工变换)和最终被处理(观察者)的发生在哪一个调度器。可指定多次,每次指定完都在下一步生效。

RxJava的线程环境有哪些选项:

RxJava原理学习_java

3.RxJava的操作符

常用的是Map和FlatMap。

①关于Map的使用例子

Observable.just(getFilePath())
//使用map操作来完成类型转换
.map(new Func1<String, Bitmap>() {
@Override
public Bitmap call(String s) {
//显然自定义的createBitmapFromPath(s)方法,是一个极其耗时的操作
return createBitmapFromPath(s);
}
})
.subscribe(
//创建观察者,作为事件传递的终点处理事件
new Subscriber<Bitmap>() {
@Override
public void onCompleted() {
Log.d("DDDDDD","结束观察...\n");
}

@Override
public void onError(Throwable e) {
//出现错误会调用这个方法
}
@Override
public void onNext(Bitmap s) {
//处理事件
showBitmap(s)
}
);

②关于FlatMap的例子

需求:查找一个学校每个班级的每个学生,并打印出来。

//创建被观察者,获取所有班级
Observable.from(getSchoolClasses())
.flatMap(new Func1<SingleClass, Observable<Student>>() {
@Override
public Observable<Student> call(SingleClass singleClass) {
//将每个班级的所有学生作为一列表包装成一列Observable<Student>,将学生一个一个传递出去
return Observable.from(singleClass.getStudents());
}
})
.subscribe(
//创建观察者,作为事件传递的终点处理事件
new Subscriber<Student>() {
@Override
public void onCompleted() {
Log.d("DDDDDD","结束观察...\n");
}

@Override
public void onError(Throwable e) {
//出现错误会调用这个方法
}
@Override
public void onNext(Student student) {
//接受到每个学生类
Log.d("DDDDDD",student.getName())
}
);

解释:FlatMap可以再次包装新的​​Observable​​​,而每个​​Observable​​​都可以使用​​from(T[])​​​方法来创建自己,这个方法接受一个列表,然后将列表中的数据包装成一系列事件。
在这里就是,​​​singleClass.getStudents()​​​返回​​Student​​​集合,然后​​Observable.from(这个Student集合)​​​,可以依次将​​Student​​​集合中的每一个元素​​Student​​​包装成​​Observable​​返回,并传递下去

4.RxJava的观察者模式

4.1 用到的类的介绍

RxJava的观察者模式是扩展的观察者模式,一个被观察者对应一个观察者。涉及的类有

Observable:被观察者
Observer:观察者
Event:被观察者通知观察者的事件
Subscribe:订阅

其中事件有

Next:常规事件,可以传递各种各样的数据;
Complete:结束事件,当观察者接收到结束事件后,就不会再接收后续被观察者发送来的事件;
Error:异常事件,当被观察者发送异常事件后,那么其他的事件就不会再继续发送了;

4.2 原理

有一个泛型类,包装了​​Observer​​​,即​​CreateEmitter​​​类。在​​Observable​​​调用​​subscribe​​​传入​​Observer​​​的时候,会首先把​​Observer​​​用​​CreateEmitter​​​包裹起来,然后当有事件发生的时候,通过这个包裹了​​Observer​​​的发射器​​CreateEmitter​​​发射事件,最终调用到​​Observer​​​里面的相应方法。
同时注意,在RxJava里面是被观察者订阅观察者。但背后还是观察者订阅的被观察者,之所以写的时候是相反的,是为了保持流失API的调用风格,以​​​Observable​​为主体,一气呵成。

​CreateEmitter​​代码如下

public class CreateEmitter<T> implements Emitter<T> {

final Observer<T> observer;

CreateEmitter(Observer<T> observer) {
this.observer = observer;
}

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

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

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

​Observable​​实现如下

public abstract class Observable<T> {

// 实现订阅的逻辑
public void subscribe(Observer<T> observer){
// 通过将传进来的observer包装成CreateEmitter,用于回调
CreateEmitter<T> emitter = new CreateEmitter<T>(observer);
// 回调订阅成功的方法
observer.onSubscribe(emitter);

// 回调发射器emitter
subscribe(emitter);
}

// 订阅成功后,进行回调
public abstract void subscribe(Emitter<T> emitter);

}

然后在​​observer​​方法中这样调用

private void observer() {
// 第一步,创建被观察者
Observable<String> observable = new Observable<String>() {
@Override
public void subscribe(Emitter<String> emitter) {
emitter.onNext("第一次");

emitter.onNext("第二次");

emitter.onNext("第三次");

emitter.onComplete();
}
};

// 第二步,创建观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Emitter emitter) {
Log.i("TAG", " onSubscribe ");
}

@Override
public void onNext(String s) {
Log.i("TAG", " onNext s:" + s);
}

@Override
public void onError(Throwable e) {
Log.i("TAG", " onError e:" + e.toString());
}

@Override
public void onComplete() {
Log.i("TAG", " onComplete ");
}
};

// 第三步,被观察者订阅观察者
observable.subscribe(observer);
}

5.RxJava的装饰者模式

我们在创建被被观察者的时候,会对被观察者做一层包装, 创建几次就包装几次,然后在被观察者调用​​subscribe​​​方法时,一层层回调被观察者的​​subscribeActual​​​方法,而在被观察者的​​subscribeActual​​​方法里,会对观察者做一层包装;
也就是说被观察者是在创建的时候进行包装,然后在​​​subscribeActual​​​方法里实现额外的功能;
而观察者是在被观察者调用​​​subscribeActual​​方法里进行包装的,然后针对观察者实现自己额外的功能;

流程图是这样的

RxJava原理学习_java_02

6.参考文章

这是​​RxJava入门学习​​​的文章推荐,这是对​​RxJava设计模式以及源码​​介绍的文章推荐。


举报

相关推荐

0 条评论