1.Rxjava的使用
关于Rxjava的使用，网上有太多太多，我这里把所有操作符汇总出来，大家有需要了解的自行查找用法，当然我也写了一点点使用方法，在文后的github链接中，这篇文章主要是想写写自己在学习Rxjava中做的笔记，希望能理解得更深刻。
Rxjava操作符汇总.png
详细的使用方法文档传送门：git@github.com:oujie123/UnderstandingOfRxJava.git
2.Rxjava思想
任何需求只有一个起点和一个终点，起点开始流向一个"事件"，"事件"经过处理之后最终流向终点，期间事件的处理可以分为很多步骤，处理的过程和终点只关心上一个过程的结果。在订阅之前的流程称之为上游，在订阅之后的流程称之为下游。
3.Rxjava的hook点
在看源码之前，多看几个api，细心的朋友都会看到在每个API返回的之后都会有RxJavaPlugins.onAssembly()方法。他的作用是什么呢？
//io.reactivex.plugins.RxJavaPlugins类
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
默认情况下onObservableAssembly为null，导致if语句不起作用。那么Rxjava设计这个有什么用呢？他的主要用途就是用于hook每一次调用。注意，这个方法会在每一个API最后中调用，意味着每一个Rxjava的方法都会经历这个方法。所以就可以通过这个方法来hook开发者的每一次调用。那么使用方法是怎样的呢？、
可以通过一下方法设置一个处理方法
public static void setOnObservableAssembly(@Nullable Function<? super Observable, ? extends Observable> onObservableAssembly) {
if (lockdown) {
throw new IllegalStateException("Plugins can't be changed anymore");
}
RxJavaPlugins.onObservableAssembly = onObservableAssembly;
}
// apply中可以加入自己想实现的代码
RxJavaPlugins.setOnObservableAssembly(new Function<io.reactivex.Observable, io.reactivex.Observable>() {
@Override
public io.reactivex.Observable apply(io.reactivex.Observable observable) throws Exception {
Log.i("jack","api of rxjava is invoked ！");
return observable;
}
});
4.Rxjava核心实现
首先列出基本代码，其他的操作符都是一样的
//返回Observable<String> 真实的是ObservaleCreate<String>()
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("JackOu");
}
})
//ObservaleCreate.map
.map(new Function<String, Bitmap>() {
@Override
public Bitmap apply(String s) throws Exception {
return null;
}
})
//ObservableMap.subscribe()
.subscribe(new Observer<Bitmap>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Bitmap bitmap) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
分析分为3步：
- 创建Observer，就是new一个接口 自定义观察者
- 创建Observable
- subscribe订阅过程
4.1 创建观察者
public interface Observer<T> {
// 当订阅成功后回调
void onSubscribe(@NonNull Disposable d);
// 开发者调用onNext回调
void onNext(@NonNull T t);
// 通知观察者发生异常时回调
void onError(@NonNull Throwable e);
// 事件处理结束回调
void onComplete();
}
观察者的创建就是创建一个回调实现，用于接收上游传下来的事件结果。
4.2 创建Observable
调用流程如下：
创建Observable.png
// 第一步调用create方法，传入自定义source
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("novel1");
}
})
//创建ObservableCreate封装自定义source
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
//最终用ObservableCreate包裹了自定义source（ObservableOnSubscribe接口）
4.3 subscribe订阅过程
订阅过程.png
// 1.当subscribe()一被调用，Observable的subscribe()方法会被调用
public final void subscribe(Observer<? super T> observer) {
......
try {
......
// 调入子类
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
......
}
// 2.调用子类ObservaleCreate的subscribeActual方法
protected void subscribeActual(Observer<? super T> observer) {
// 创建发射器包裹自定义的观察者
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
// 调用观察者的onSubscribe订阅回调
observer.onSubscribe(parent);
try {
// 上述创建的自定义source，即ObservableOnSubscribe
// 调用ObservableOnSubscribe的subscribe传入自定义观察者
// 开发者调用发射器的onNext()就会调到自定义观察者的onNext()中
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
4.4 原理图
Rxjava原理图.png
5.标准观察者模式和Rxjava的变异观察者模式有哪些区别
在标准的观察者模式，是一个“被观察者”，多个“观察者”，当被观察者有事件通知的时候，所有观察者都会收到；在标准的观察者设计模式中：当发出通知改变时，会遍历Observable里面的容器，此容器里面有10个Observer，就会通知10个Observer
rxjava的观察者模式是多个“被观察者”，一个“观察者”，需要起点和终点订阅一次之后，但发出改变通知，观察者才能看到。在RxJava观察者设计模式中：分发事件时，会拿到发射器，通过发射器关联到我们自定义的Observer，发射器调用到我们自定义的Observer
6.在创建类操作符中create和just的区别
just内部封装了一个run方法，自动执行了，observer.onNext()
create需要程序员去调用createEmitter.onNext（）
网友评论