Dieser Artikel beschreibt die grundlegende Verwendung von RxJava. RxJava ist ein magisches Framework mit sehr einfacher Verwendung, aber die interne Implementierung ist etwas kompliziert und die Codelogik ist etwas kompliziert. Es gibt nur wenige Online-Artikel zur RxJava-Quellcode-Analyse und unvollständige Quellcode-Beiträge. Die vollständige Quellcode-Analyse ist unten als Referenz aufgeführt.
1. Grundlegende Verwendung von RxJava
Observable.create(new Observable.OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
}
}).subscribe(new Observer<Object>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
}
});2. Schauen Sie sich zuerst die Methode .subscribe(new Observer
public final Subscription subscribe(final Observer<? super T> observer) {
if (observer instanceof Subscriber) {
return subscribe((Subscriber<? super T>)observer);
}
if (observer == null) {
throw new NullPointerException("observer is null");
}
return subscribe(new ObserverSubscriber<T>(observer));
}Dies ist nur der Beobachterparameter das wird übergeben Eine einfache Kapselung (ObserverableSubscriber)
Schauen Sie sich weiterhin die Abonnementmethode an
public final Subscription subscribe(Subscriber<? super T> subscriber) {
return Observable.subscribe(subscriber, this);
} static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
// validate and proceed
if (subscriber == null) {
throw new IllegalArgumentException("subscriber can not be null");
}
if (observable.onSubscribe == null) {
throw new IllegalStateException("onSubscribe function can not be null.");
/*
* the subscribe function can also be overridden but generally that's not the appropriate approach
* so I won't mention that in the exception
*/
}
// new Subscriber so onStart it
subscriber.onStart();
/*
* See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
* to user code from within an Observer"
*/
// if not already wrapped
if (!(subscriber instanceof SafeSubscriber)) {
// assign to `observer` so we return the protected version
subscriber = new SafeSubscriber<T>(subscriber);
}
// The code below is exactly the same an unsafeSubscribe but not used because it would
// add a significant depth to already huge call stacks.
try {
// allow the hook to intercept and/or decorate
RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
return RxJavaHooks.onObservableReturn(subscriber);
} catch (Throwable e) {
// special handling for certain Throwable/Error/Exception types
Exceptions.throwIfFatal(e);
// in case the subscriber can't listen to exceptions anymore
if (subscriber.isUnsubscribed()) {
RxJavaHooks.onError(RxJavaHooks.onObservableError(e));
} else {
// if an unhandled error occurs executing the onSubscribe we will propagate it
try {
subscriber.onError(RxJavaHooks.onObservableError(e));
} catch (Throwable e2) {
Exceptions.throwIfFatal(e2);
// if this happens it means the onError itself failed (perhaps an invalid function implementation)
// so we are unable to propagate the error correctly and will just throw
RuntimeException r = new OnErrorFailedException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
// TODO could the hook be the cause of the error in the on error handling.
RxJavaHooks.onObservableError(r);
// TODO why aren't we throwing the hook's return value.
throw r; // NOPMD
}
}
return Subscriptions.unsubscribed();
}
}Sie müssen hier nur aufpassen
RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
public static <T> Observable.OnSubscribe<T> onObservableStart(Observable<T> instance, Observable.OnSubscribe<T> onSubscribe) {
Func2<Observable, Observable.OnSubscribe, Observable.OnSubscribe> f = onObservableStart;
if (f != null) {
return f.call(instance, onSubscribe);
}
return onSubscribe;
}RxJavaHooks.onObservableStart(observable, observable.onSubscribe) 这个方法返回的是它的第二个参数，也就是Observable它自己的onSubscribe 对象， 所以在subscribe 方法里面调用了 onSubscribe.call(subscriber)方法
Der Abonnent hier ist der übergebener Parameter
protected Observable(OnSubscribe<T> f) {
this.onSubscribe = f;
}public static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(RxJavaHooks.onCreate(f));
}Es ist ersichtlich, dass das onSubscribe-Objekt der von create übergebene Parameter ist, dann ist der gesamte Prozess sehr klar
Der gesamte Prozess wird nur ausgeführt, wenn die subscribe-Methode ausgeführt wird heißt: subscribe ===>Call onSubscribe.call(observer) Methode übergibt auch den Beobachter in
Verwandte Empfehlungen:
RxJava Operator (8) Aggregate_PHP Tutorial
Video: JavaScript Basic-Video-Tutorial zur Stärkung
Das obige ist der detaillierte Inhalt vonGrundlegende RxJava-Anwendungsfall-Sharing-Quellcode-Analyse [mit Code]. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
![[Web-Frontend] Node.js-Schnellstart](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
