基于
RxJava版本2.2.2分析
Hello World?
private void doSomeWork() {
getObservable().subscribe(getObserver());
}
/**
*
* @return 上游
*/
private Observable<String> getObservable() {
return Observable.just("Cricket", "Football");
}
/**
*
* @return 订阅的下游
*/
private Observer<String> getObserver() {
return new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, " onSubscribe : " + d.isDisposed());
}
@Override
public void onNext(String value) {
textView.append(" onNext : value : " + value);
textView.append(AppConstant.LINE_SEPARATOR);
Log.d(TAG, " onNext : value : " + value);
}
@Override
public void onError(Throwable e) {
textView.append(" onError : " + e.getMessage());
textView.append(AppConstant.LINE_SEPARATOR);
Log.d(TAG, " onError : " + e.getMessage());
}
@Override
public void onComplete() {
textView.append(" onComplete");
textView.append(AppConstant.LINE_SEPARATOR);
Log.d(TAG, " onComplete");
}
};
}
上面是对RxJava的简单使用，创建上游然后下游订阅。这里分为三部分，上游、订阅以及下游，我们一步一步的分析。
上游
private Observable<String> getObservable() {
return Observable.just("Cricket", "Football");
}
从Observable.just(T,T)中，不难猜到，这是一个静态方法，而且该方法的返回值是一个Observable<T>，在RxJava中的源码是这样实现的：
public static <T> Observable<T> just(T item1, T item2) {
ObjectHelper.requireNonNull(item1, "The first item is null");
ObjectHelper.requireNonNull(item2, "The second item is null");
return fromArray(item1, item2);
}
public static <T> Observable<T> fromArray(T... items) {
ObjectHelper.requireNonNull(items, "items is null");
if (items.length == 0) {
return empty();
} else
if (items.length == 1) {
return just(items[0]);
}
return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));
}
和我们猜想的差不多，Observable.just(T,T)方法最终调用RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));实现，该方法返回一个Observable<T>，那么看一下RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));是如何实现的。
Assembly有装配的意思
@NonNull
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
//在我们的例子中，f==null 为 true，所以直接就放回source
return source;
}
在我们的例子中在我们的例子中，f == null为true，所以直接就返回source,也就是说，直接返回RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));中的new ObservableFromArray<T>(items)，这样很自然能可以猜到ObservableFromArray是Observable的其中一个实现类。
public final class ObservableFromArray<T> extends Observable<T> {
final T[] array;
public ObservableFromArray(T[] array) {
this.array = array;
}
......
}
到这里，我们知道上游返回的是 Observable的一个实现类ObservableFromArray，接下来我们看订阅，到最后再把所有串联起来。
订阅
private void doSomeWork() {
getObservable().subscribe(getObserver());
}
还记得我们getObservable()返回的是什么吗？返回的是Observable，那么subscribe肯定是Observable的方法
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
这里重点关注subscribeActual(observer);,因为其他方法都是判断是否为空，RxJavaPlugins.onSubscribe(this, observer);在我们的例子里面，是原封不动的返回observer
protected abstract void subscribeActual(Observer<? super T> observer);
subscribeActual是一个抽象的方法，具体实现在ObservableFromArray中，
@Override
public void subscribeActual(Observer<? super T> observer) {
FromArrayDisposable<T> d = new FromArrayDisposable<T>(observer, array);
observer.onSubscribe(d);
if (d.fusionMode) {
return;
}
d.run();
}
FromArrayDisposable是一个数组处理者，observer.onSubscribe(d);,这里将会回调下游的onSubscribe方法，在我们的例子中d.fusionMode返回的是false所以忽略，我们直接看d.run();
void run() {
T[] a = array;
int n = a.length;
for (int i = 0; i < n && !isDisposed(); i++) {
T value = a[i];
if (value == null) {
downstream.onError(new NullPointerException("The " + i + "th element is null"));
return;
}
downstream.onNext(value);
}
if (!isDisposed()) {
downstream.onComplete();
}
}
上面的代码很简单，就是遍历数组和判断是否处理了，然后回调给下游(这里的downstream就是observer)。
到此HelloWorld的代码就分析的差不多了，你可以在GitHub找到我仿照RxJava流程写的伪代码。
