android Rxjava：实现原理浅析

从使用步骤逐步深入：

使用步骤

Observable.create(new ObservableOnSubscribe() {

@Override

public void subscribe(@NonNull ObservableEmitter emitter) throws Throwable {

emitter.onNext(1);

emitter.onNext(2);

emitter.onNext(3);

emitter.onComplete();

}

}).subscribe(new Observer() {

@Override

public void onSubscribe(@NonNull Disposable d) {

Log.d("main", "开始采用subscribe连接");

}

@Override

public void onNext(@NonNull Integer integer) {

Log.d("main", "next" + integer);

}

@Override

public void onError(@NonNull Throwable e) {

Log.d("main", "对Error事件作出响应");

}

@Override

public void onComplete() {

Log.d("main", "onComplete");

}

});

被观察者创建过程

public static <@NonNull T> Observable create(@NonNull ObservableOnSubscribe source) {

Objects.requireNonNull(source, "source is null");

return RxJavaPlugins.onAssembly(new ObservableCreate<>(source));

}

核心是new ObservableCreate<>(source)，传入的相当于事件发出的source

public final class ObservableCreate extends Observable {

final ObservableOnSubscribe source;

public ObservableCreate(ObservableOnSubscribe source) {

this.source = source;

}

@Override

protected void subscribeActual(Observer observer) {

CreateEmitter parent = new CreateEmitter<>(observer); //用于发射事件

observer.onSubscribe(parent); //调用观察者中重写的方法onSubscribe

try {

source.subscribe(parent); //调用被观察者的subscribe方法（产生事件）

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

parent.onError(ex);

}

}

该类内部把source存储起来，并且重写的subscribeActual方法将observer观察者和observable被观察者联系起来，用于触发。

事件onNext

@Override

public void onNext(T t) {

if (t == null) {

onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));

return;

}

if (!isDisposed()) {

observer.onNext(t);

}

}

onNext的实现如上，若没有disposed（关闭连接），就执行onNext。（此处在create内只是定义好方法，并没有真正执行，只有subscribe触发subscribeActual，该方法里面调用了source.subscribe(parent)就是真正执行这些定义好的方法）

观察者创建过程

new Observer()

Observer本身就是一个接口，里面四个方法响应被观察者的不同事件

观察者与被观察者的联系subscribe

public final void subscribe(@NonNull Observer observer) {

Objects.requireNonNull(observer, "observer is null");

try {

observer = RxJavaPlugins.onSubscribe(this, observer);

subscribeActual(observer);

}

这里就发现subscribe会调用内部的subscribeActual，就会再调用被观察者实现的subscribe方法，其中的onNext等内容就会被调用

文章来源

 您阅读本篇文章共花了：