admin管理员组文章数量:1632798
从Demo到原理
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d("MainActivity:","onSubscribe()");
}
@Override
public void onNext(Integer i) {
Log.d("MainActivity : " ,"onNext =" + i);
}
@Override
public void onError(Throwable e) {
Log.d("MainActivity:","onError()");
}
@Override
public void onComplete() {
Log.d("MainActivity:","onComplete()");
}
});
输出结果:
MainActivity:: onSubscribe()
MainActivity :: onNext =1
MainActivity :: onNext =2
MainActivity :: onNext =3
MainActivity:: onComplete()
可以看到,Observer的onSubscribe是最先被调用的,这个回调会有什么用呢?我们后面会讲到。
OK,从哪开始入手呢?Observable.create,嗯,整个流程是从create开始的,那么我们就从源头开始吧。先看一下create,他返回的是一个observable对象,也就是被观察的对象。create方法需要传入一个ObservableOnSubscribe来创建,我们看下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(ObservableEmitter<T> e) throws Exception;
}
很显然,ObservableOnSubscribe是一个接口,里面只有一个方法subscribe(),我们来看看这个方法里的参数类型ObservableEmitter:
public interface ObservableEmitter<T> extends Emitter<T> {
```java
/**
* Sets a Disposable on this emitter; any previous Disposable
* or Cancellation will be unsubscribed/cancelled.
* @param d the disposable, null is allowed
*/
void setDisposable(Disposable d);
/**
* Sets a Cancellable on this emitter; any previous Disposable
* or Cancellation will be unsubscribed/cancelled.
* @param c the cancellable resource, null is allowed
*/
void setCancellable(Cancellable c);
/**
* Returns true if the downstream disposed the sequence.
* @return true if the downstream disposed the sequence
*/
boolean isDisposed();
/**
* Ensures that calls to onNext, onError and onComplete are properly serialized.
* @return the serialized ObservableEmitter
*/
ObservableEmitter<T> serialize();
}
public interface Emitter<T> {
/**
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(@NonNull T value);
/**
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(@NonNull Throwable error);
/**
* Signal a completion.
*/
void onComplete();
}
ObservableEmitter继承了Emitter这个类,而却它还多了几个方法,这里其实就是Rxjava 2.0之后引入的功能,提供了可以中途取消等新能力,我们继续看Emitter这个类,相信大家对这三个方法应该很熟悉了,这里就不过多介绍了。
ObservableOnSubscribe 是这个事件流的源头,下面我们称之为事件源,一般由我们自己创建并传入。我们创建时,需要重写其subscribe()方法,通过传入的ObservableEmitter参数,进去事件的生产。看到这里,我们只是了解了传递参数的数据结构,了解到的信息还是比较少的。我们继续看下create内部做了什么操作呢?
Class Observable{
......
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
......
}
这里的RxJavaPlugins.onAssembly运用的是Hook技术,这里大家就理解成返回一个Observable对象就可以了。我在主要来看看
ObservableCreate这个类:
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
@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);
}
}
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();
}
}
}
ObservableCreate继承自Observable,这里的source就是上文提到的ObservableOnSubscribe,也就是Create()方法传入的参数:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(123);
}
})
我们重点来看一下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);
}
}
这里传入了一个observer(观察者)作为参数再次传入到CreateEmitter这个内部类中,我们跟着代码再来看看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();
}
}
}
相信到这里,大家就基本能看懂了,没错,这里就是我们输入事件的地方,这里根据传入的observer参数,比如调用onNext()方法是,同时也会调用Observer的onNext()方法,起到事件监听的效果。同时CreateEmitter是实现了Disposable 的,这个类只要是起到中断事件传递的作用,通过isDisposed()的布尔值,来决定是继续分发监听事件还是就此调用onComplete()结束事件,同时它还保证了onComplete()和onError()不能同时都调用的情况。我们接着往下看代码:
source.subscribe(parent);
这个将CreateEmitter作为参数,调用了ObservableOnSubscribe里的subscribe方法,也就是执行这里:
综上所述,可以说subscribeActual()这个方法起到了承上启下的作用,是Rxjava的核心所在!那么这个方法具体是在那里调用呢?我们继续跟随源码分析:
这里调用了Observerable的subscribe()方法:
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {}
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
subscribeActual(observer);
}
......
}
这里传入了observer对象,同时调用了oberver的onSubscribe()方法,这也是为什么onSubscribe()之所以第一次调用的原因,接着就调用了我们最重要的subscribeActual(observer)方法了,从这里也可以知道,整个Rxjava的调用链中,只有subscribe(订阅)发生了,才会进行事件的监听和分发!
版权声明:本文标题:Rxjava 2 的源码解析 内容由热心网友自发贡献,该文观点仅代表作者本人, 转载请联系作者并注明出处:https://www.elefans.com/xitong/1729149229a1187902.html, 本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容,一经查实,本站将立刻删除。
发表评论