前言
看完RxJava2的源码，真是惊艳。代码好懂，流程清晰。
本文是对RxJava2的抽象，自己写了一个RxJava来理解它的设计和流程，实现了几个操作符、运用了简单工厂模式、观察者模式、责任链模式、装饰者模式等，理解链式调用的设计、加深了对函数式编程和响应式编程的理解。
分为如下几个方面去理解，将会用代码说明：
1. 被观察者和观察者是如何关联起来的
// 被观察者，数据发送者
interface Emitter<T> {
void onNext(T t);
void onComplete();
void onError(Throwable e);
}
// 数据接收者，观察者
interface Observer<T> extends Emitter<T> {
// 观察上了
void onSubscription();
}
// 关联接口，就是通过产生关联
interface ObservableOnSubscribe<T> {
void subscribe(Emitter<T> emitter);
}
class Observable<T> {
private ObservableOnSubscribe<T> mObservableOnSubscribe;
private Observable(ObservableOnSubscribe<T> observableOnSubscribe) {
this.mObservableOnSubscribe = observableOnSubscribe;
}
// 创建型操符
static <T> Observable<T> create(ObservableOnSubscribe<T> observableOnSubscribe) {
return new Observable<>(observableOnSubscribe);
}
void subscribe(final Observer<T> observer) {
observer.onSubscription(); // 观察上了，运行在被观察者线程
// // 调用被观察者方法，让它产生数据，即已经观察上了，被观察者可以发送数据，观察者响应。
mObservableOnSubscribe.subscribe(observer);
}
}
从抽象代码中可以看出，是让Observer继承了Emitter，即观察者继承了被观察者，从设计上也可以理解成：自己在观察自己。所以自己在发送数据当然自己能接收到。
但是也可以从语义上理解，就是被观察者发送数据、观察者响应。
在RxJava2代码中，Observer没有继承Emitter，即从设计上就严格区分了这两者。
2. map操作符如何起作用
<R> Observable<R> map(final Function<T, R> function) {
// 会返回一个新的Observable，持有新的mObservableOnSubscribe，可以被下游调用
return new Observable<>(new ObservableOnSubscribe<R>() {
@Override
public void subscribe(final Emitter<R> emitter) {
// 观察者会调用这块，新包装的ObservableMap会调用上游的subscribe
// 传入新的emitter被上游调用，自己处理
mObservableOnSubscribe.subscribe(new Emitter<T>() {
@Override
public void onNext(T t) {
emitter.onNext(function.apply(t));
}
@Override
public void onComplete() {
emitter.onComplete();
}
@Override
public void onError(Throwable e) {
emitter.onError(e);
}
});
}
});
}
可以看到，使用装饰者模式实现map操作符这个功能，调用function增强了功能。
另外在附上filter操作符的实现，加深理解。
Observable<T> filter(final Prediction<T> prediction) {
return new Observable<>(new ObservableOnSubscribe<T>() {
@Override
public void subscribe(final Emitter<T> emitter) {
mObservableOnSubscribe.subscribe(new Emitter<T>() {
@Override
public void onNext(T t) {
if (prediction.filter(t)) {
emitter.onNext(t);
}
}
@Override
public void onComplete() {
emitter.onComplete();
}
@Override
public void onError(Throwable e) {
emitter.onError(e);
}
});
}
});
}
3. 事件、数据是怎么传递的
首先，得理解它是一系列的装饰者模式，或者代理模式吧，不同的操作符代表不同的增强类。很类似Java里的IO类吧，一层一层包裹，不断增强功能。不过Rx里的操作符可以控制数据，是代理模式吧。
在调用操作符时，会把上游被观察传入进去，在调用subscribe时，是调用的上游的被观察者的subcribe方法，然后在onNext里上游观察者会向下传递，是因为传入了自己的ObservableOnSubscribe，上游会调用该emitter。
调用每个操作符会创建新的Observable，该Observable都会持有上游的Observable，来传递subscribe的调用；每个Observable都有自己的观察者，同时将自己的观察者传给上游，即上游被下游的观察者观察，这样就完成了数据的链式调用。
如图所示
4. 如何切换线程的
切换线程demo这写的比较简单，subscribeOn指定上游发生线程，observeOn指定下游发生线程。即：
Observable<T> subscribeOn(final Scheduler scheduler) {
return new Observable<>(new ObservableOnSubscribe<T>() {
@Override
public void subscribe(final Emitter<T> emitter) {
scheduler.schedule(new Runnable() {
@Override
public void run() {
// 让上游的调用发生在指定线程里
mObservableOnSubscribe.subscribe(emitter);
}
});
}
});
}
Observable<T> observeOn(final Scheduler scheduler) {
return new Observable<>(new ObservableOnSubscribe<T>() {
@Override
public void subscribe(final Emitter<T> emitter) {
mObservableOnSubscribe.subscribe(new Emitter<T>() {
@Override
public void onNext(final T t) {
// 下游发生在指定线程里
scheduler.schedule(new Runnable() {
@Override
public void run() {
emitter.onNext(t);
}
});
}
@Override
public void onComplete() {
scheduler.schedule(new Runnable() {
@Override
public void run() {
emitter.onComplete();
}
});
}
@Override
public void onError(final Throwable e) {
scheduler.schedule(new Runnable() {
@Override
public void run() {
emitter.onError(e);
}
});
}
});
}
});
}
5. Flowable、背压的实现
在RxJava2中，Flowable和Observable是两套代码。Flowable也对应着自己的一套接口，关于背压的实现先留着，之后再写。
结语
本文实现了简单的RxJava版本，实现了类似如下代码的调用，实现过程中对Rx的理解也更深，对设计模式的使用也明白更多。毕竟抽象一下自己竟然可以观察自己，自己调用自己方法当然能被监听，但实现很巧妙，中间可以操作数据，普通的写法不会这么优美。
之后在生产实践中也可以参考这一套，去看看哪些场景、代码设计可以用到这些。观察者模式是不是可以写的更优美些呢。
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(Emitter<String> e) {
System.out.println("create: " + Thread.currentThread().getName());
e.onNext("1");
e.onNext("2");
e.onNext("3");
e.onNext("4");
e.onComplete();
}
})
.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) {
System.out.println("apply: " + Thread.currentThread().getName());
int i = Integer.parseInt(s);
return i * i;
}
})
.filter(new Prediction<Integer>() {
@Override
public boolean filter(Integer integer) {
System.out.println("filter: " + integer + ", " + Thread.currentThread().getName());
return integer % 2 == 0;
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.io())
.subscribe(new Observer<Integer>() {
@Override
public void onSubscription() {
System.out.println("onSubscription: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer s) {
System.out.println("onNext:" + s + ", " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
System.out.println("onComplete: " + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
System.out.println("onError: " + e.toString());
}
});
网友评论