要用RxJava,那么就得先了解两个概念
Observable被观察者
Observer观察者
比如放羊,羊是被观察者,放羊娃就是观察者,当羊饿了的时候,咩咩的叫个不停,相当于向放羊娃发出信号说我饿了,你看着办,放羊娃呢,在听到羊的叫声后就会做出相应的处理.那么这里前提是二者要关联成为观察与被观察的关系.否则没有关联的关系,放羊娃才不会那么傻去喂养和自己没关系的羊.
观实者与被观察者就是这样的状态,本来各不相干,但是一旦二者关联,那么观察者就和被观察者形成关系,观察者会对被观察者的言行举止做出相应的响应.
Observable被观察者(小羊)
首先是被观察者的创建:(羊来了)
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
// 羊咩咩的叫个不停于是用e向外发射信号
e.onNext("我饿了");
}
});
此处有两个概念:
ObservableOnSubscribe : 一个接口,创建被观察者时需要传入他的实例
ObservableEmitter : 事件发射器
create方法做了判空处理后,返回了ObservableCreate的实例
ObservableCreate是Observable的子类实现
Observer 观察者(放羊娃)
观察者的创建
Observer observer = new Observer() {
@Override
public void onSubscribe(Disposable d) { }
@Override
public void onNext(String s) {
// 接收到羊发来的信号s,我知道啦,马上给你准备草去
}
@Override
public void onError(Throwable e) {}
@Override
public void onComplete() { }
};
Observer用来对被观察者发射出的事件做相应的处理,也可理解为对监听到被观察者的相应的状态做出相应的处理.
observable.subscribe(observer);
最后通过subscribe方法完成被观察者与观察者的订阅连接 ,此时羊与放羊娃明确关系.
subscribe方法内部实现其实最终调用的是observable的subscribeActual 实现方法
@Override
protected void subscribeActual(Observer observer) {
CreateEmitter parent = new CreateEmitter(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
可以看到,该方法会将我们传入的observer再次传给CreateEmitter, CreateEmitter是ObservableCreater 的一个内部类,同时实现了ObservableEmitter,Disposable这两个接口,相当于关联了Observer的基础上又做了进一步的封装,因为其内部实现最终事件的执行都是通过传入的observer来完成的
其持有了observer实例后,在可以执行observer的相应事件方法的基础上进一步的添加了一些控制订阅,以及状态判断的方法.如下源码可见:
static final class CreateEmitter
extends AtomicReference
implements ObservableEmitter, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer observer;
CreateEmitter(Observer 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 (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(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();
}
return true;
}
return false;
}
@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 serialize() {
return new SerializedEmitter(this);
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}
ObservableEmitter也是一个接口,是Emitter的子类,如下源码:
public interface ObservableEmitter extends Emitter
public interface Emitter {
/**
* 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();
}
***********************************************************
public interface Disposable {
/**
* Dispose the resource, the operation should be idempotent.
*/
void dispose();
/**
* Returns true if this resource has been disposed.
* @return true if this resource has been disposed
*/
boolean isDisposed();
}
onSubscribe方法将观察者和emmitter发射器关联起来,
往下走source.subscribe(parent);
此处的source就是我们在创建被观察者是传入的(new ObservableOnSubscribe())
至此可以看到ObservableOnSubscribe实例的subscriber方法将自身与关联了Observer的事件发射器进行了关联.
总结
1.被观察者Observabler的创建最终返回的是ObservableCreate的实例,所以被观察者Observabler的状态以及事件的执行都是通过ObservableCreate的内部类CreateEmitter来实现的,
2.CreateEmitter对订阅的Observer进行了关联封装
3.被观察者Observable通过subscribe方法将观察者传给他的内部类CreateEmitter用来进行关联封装.
4.ObservableCreate创建的时候接收了Observable.create方法传入的(newObservableOnSubscribe())并与自身的内部类CreateEmitter进行了关联
5.被观察者Observabler要执行的事件都会在传入的ObservableOnSubscribe实例的subscriber方法中进行操作再经由关联的CreateEmitter向外发射.最终由CreateEmitter关联封装的Observer接收事件并作出响应.
网友评论