RxJava2.0中create操作符用法和源码分析
程序员文章站
2024-02-29 08:28:46
...
create基本使用
create是用来从头开始创建一个Observable的对象，应该调用ObservableEmitter的onNext, onError 和onComplete方法,如下：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//Emitter data
}
})create接收的参数是ObservableOnSubscribe对象，查看源码：
public interface ObservableOnSubscribe<T> {
/**
* Called for each Observer that subscribes.
* @param e the safe emitter instance, never null
* @throws Exception on error
*/
void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;
}我们再继续分析create方法:
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}将传入的ObservableOnSubscribe对象作为参数来生成一个ObservableCreate对象,最后调用RxJavaPlugins类中的onAssembly方法来生成一个Observable对象。
@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);
}
return source;
}通过onAssembly方法可知，直接返回ObservableCreate对象，由此可知ObservableCreate是Observable的子类。如下：
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
....
}完成了Observable被观察者的创建，然后开始订阅Observer观察者，这中间通过关键的方法subscribe()来完成调用。这之间是如何进行协同工作的，我们同样通过源码来分析。我们先来分析subscribe方法，如下:
@Override
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
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对象关联，当我们查看subscribeActual方法时，发现是个抽象的方法：
protected abstract void subscribeActual(Observer<? super T> observer);那么这个方法具体是由谁执行的呢，在前面我们知道通过create方法，我们创建了一个Observable对象，而这个对象是ObservableCreate对象。由此可知，调用的方法应该是该类中的subscribeActual方法。
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
如上，首先会去创建一个CreateEmitter对象，然后将CreateEmitter作为参数回调observer中的onSubscribe方法。
我们再来看看CreateEmitter类：
static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
} else {
RxJavaPlugins.onError(t);
}
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
@Override
public void setDisposable(Disposable d) {
DisposableHelper.set(this, d);
}
@Override
public void setCancellable(Cancellable c) {
setDisposable(new CancellableDisposable(c));
}
@Override
public ObservableEmitter<T> serialize() {
return new SerializedEmitter<T>(this);
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}从上面我们可以看到，CreateEmitter是继承自ObservableEmitter类，我们也终于在类型看到了熟悉的方法onNext，onError，onComplete，当我们通过ObservableEmitter调用onNext，onError，onComplete方法时，这时候都会有对应的Observer中的onNext，onError，onComplete方法进行回调。
所以说只有在Observable完成了subscribe订阅之后，才能对ObservableOnSubscribe中subscribe方法的调用，进而使用ObservableEmitter调用onNext，onError，onComplete等方法。
下面列出create操作符的简单demo：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
println("Observable emit 1" + "\n");
e.onNext(1);
println("Observable emit 2" + "\n");
e.onNext(2);
println("Observable emit 3" + "\n");
e.onNext(3);
e.onComplete();
println("Observable emit 4" + "\n");
e.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDisposable;
@Override
public void onSubscribe(Disposable d) {
println("onSubscribe : " + d.isDisposed() + "\n");
mDisposable = d;
}
@Override
public void onNext(Integer integer) {
println("onNext : value : " + integer + "\n");
if (integer == 2) {
// 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
mDisposable.dispose();
println("onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
}
}
@Override
public void onError(Throwable e) {
println("onError : value : " + e.getMessage() + "\n");
}
@Override
public void onComplete() {
println("onComplete" + "\n");
}
});
输出结果：
onSubscribe : false
Observable emit 1
onNext : value : 1
Observable emit 2
onNext : value : 2
onNext : isDisposable : true
Observable emit 3
Observable emit 4