Rxjava2 Observable源码浅析
- 上游：一般来说是
Observable、Flowable、Subject - 下游：一般来说是
Observer、Subscrption - cold Observable：只有通过
Observable#subscribe才开始请求上游发送数据。当下游请求dispose()停止通知上游停止发送。 - hot Observable：不管有无下游，都可以进行数据的发送
- 从
Rxjava1开始就有人说Rxjava可以看作流水线，上游怎么加工对于下游来说是无感知的，下游只要负责接收响应对应数据事件就行。
对于rxajva的思考，可以参考一下：Rxjava沉思录系列和 Rxjava主要负责人系列博客
cold Observable
一般cold Observable创建都是通过 just 、 create 、 fromXX 、 just 创建的。最简单粗暴的创建方式：
Observable.create<String> { it.onNext("") }.subscribe()
复制代码
//[仅关注点相关代码]
//ObservableOnSubscribe仅一个subscribe方法
public interface ObservableOnSubscribe<T> {
void subscribe(ObservableEmitter<T> e) throws Exception;
}
public abstract class Observable<T> implements ObservableSource<T> {
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
//RxJavaPlugins这是一个全局Hook，#onAssembly不实现默认直接返回
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
public final void subscribe(Observer<? super T> observer) {
try {
........
//真正调用subscribe的实现
subscribeActual(observer);
}
......
}
//整个Observable唯一的抽象方法，由子Observable实现，通过这个方法将上游和下游关联起来
protected abstract void subscribeActual(Observer<? super T> observer);
}
复制代码
由 Observable#create 真正返回的是 ObservableCreate ,当调用 Observable#subscribe 才真正通知上游 Observable 开始发送数据。其实质是通过 #subscribeActual 将上下游建立联系,并调用上游 #subscribe （在 ObservableCreate 中就是 ObservableOnSubscribe#subscribe ）方法通知上游，下游已订阅可以开始发送数据。
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(上游)即Observable#create传入的ObservableOnSubscribe
//这里就将上下游真正的联系了起来。
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
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);
}
}
.....
}
}
复制代码
所以实质就是下游通知上游，下游已产生订阅触发上游下发数据/事件，上游再通过下发数据/事件，最终下游通过指定方法响应上游下发的数据/事件。所以一开始说的流水线方式就可以理解了。
因为每次下游产生一次订阅都会通知到上游的 #subscribe ，所以如果上游只在 #subscribe 中去创建初始数据源就可以每个做到不同下游的数据不关联
Observable.create<String> { it.onNext("") }.subscribe() 流程图如下：
map操作符
//Observable#map
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
//**ObservableMap将上游Observable和当前的转换Function建立联系
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
//ObservableMap.java
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
//将下游包装成MapObserver，并将MapObserver和上游建立联系
//这样上游下发时，先通过MapObserver处理才下发给真正的Observer
source.subscribe(new MapObserver<T, U>(t, function));
}
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
....
U v;
try {
//通过Function获取到map后的数据
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch ....
//向下游下发数据
actual.onNext(v);
}
...
}
}
复制代码
可以看到 map 操作符的作用就是通过将上游拦截返回 ObservableMap 提供给下游订阅，并在map上游返回数据前通过 mapper 将上游数据转化并下发给下游。
线程调度
subscribeOn
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
//emmmm,是不是点眼熟
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
//ObservableSubscribeOn
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
//onSubscribe()方法执行在 订阅处所在的线程
s.onSubscribe(parent);
//将上游放入scheduler中调用,且立即执行
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer<? super T> actual;
final AtomicReference<Disposable> s;
SubscribeOnObserver(Observer<? super T> actual) {
this.actual = actual;
this.s = new AtomicReference<Disposable>();
}
@Override
public void onSubscribe(Disposable s) {
DisposableHelper.setOnce(this.s, s);
}
@Override
public void onNext(T t) {
actual.onNext(t);
}
@Override
public void onError(Throwable t) {
actual.onError(t);
}
@Override
public void onComplete() {
actual.onComplete();
}
@Override
//scheduler#scheduleDirect中执行完后
public void dispose() {
DisposableHelper.dispose(s);
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
void setDisposable(Disposable d) {
DisposableHelper.setOnce(this, d);
}
}
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//该方法调用已经在scheduler中调用
source.subscribe(parent);
}
}
}
复制代码
由源码可以看出由 scheduler.scheduleDirect -> SubscribeTask#run -> SubscribeOnObserver#subscribe(observer) 将整个调度切换到指定线程中。
因为订阅是用下自上的，所以 subscribeOn 也总是离源最近的一个生效。因为触发源的 subscribe 是离源最近一个。
observeOn
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
复制代码
可以看出Rxjava的操作符套路基本是将源 Observable 通过装饰者模式封装一层再返回新的Observable
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;//默认false
final int bufferSize;//一般128
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
//创建主线程调度器
Scheduler.Worker w = scheduler.createWorker();
//关联上下游，触发上游订阅过程
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
}
复制代码
这里可以看出 ObservableObserveOn 还是很简单的，上游订阅过程并不用关心，下游的触发则由 ObserveOnObserver 处理。
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
private static final long serialVersionUID = 6576896619930983584L;
final Observer<? super T> actual;
final Scheduler.Worker worker;
final boolean delayError;
final int bufferSize;
//上游数据的缓存队列
SimpleQueue<T> queue;
Disposable s;
Throwable error;
volatile boolean done;
volatile boolean cancelled;
int sourceMode;
boolean outputFused;
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.actual = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
......
//创建对接缓存数据
queue = new SpscLinkedArrayQueue<T>(bufferSize);
//回调下游onSubscribe
actual.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {//执行过complete/error则done为true
return;
}
if (sourceMode != QueueDisposable.ASYNC) {//非异步数据，默认同步数据
queue.offer(t);//入队列
}
schedule();
}
@Override
public void onError(Throwable t) {
if (done) {
RxJavaPlugins.onError(t);
return;
}
error = t;
done = true;//标记已完成
schedule();
}
@Override
public void onComplete() {
if (done) {
return;
}
done = true;//标记已完成
schedule();
}
@Override
public void dispose() {
if (!cancelled) {
cancelled = true;
s.dispose();
worker.dispose();
if (getAndIncrement() == 0) {
queue.clear();
}
}
}
@Override
public boolean isDisposed() {
return cancelled;
}
void schedule() {
//自旋+1，!=0则表示worker.schedule已在执行无需在调度
if (getAndIncrement() == 0) {
worker.schedule(this);//通过调度器处理，将数据取出下发到下游
}
}
@Override
public void run() {
if (outputFused) {//默认false
drainFused();
} else {
drainNormal();//取出数据下发
}
}
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (; ; ) {
//检测是否不用再处理
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (; ; ) {
boolean d = done;
T v;
try {
v = q.poll();//取出一个数据
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {//可能已经提前disposed了
return;
}
if (empty) {//数据为空队列无数据，退出下发循环
break;
}
//下发
a.onNext(v);
}
//可能有错过的schedule，再次循环检测
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
//检测是否compelte/error/队列已空
boolean checkTerminated(boolean d, boolean empty, Observer<? super T> a) {
if (cancelled) {//已经disposed
queue.clear();
return true;
}
if (d) {//是否已结束
Throwable e = error;
if (delayError) {//延迟error，等待队列清空
if (empty) {
if (e != null) {
a.onError(e);
} else {
a.onComplete();
}
worker.dispose();
return true;
}
} else {
if (e != null) {
queue.clear();
a.onError(e);//下发error
worker.dispose();
return true;
} else if (empty) {
a.onComplete();
worker.dispose();
return true;
}
}
}
return false;
}
}
复制代码
ObserveOnObserver 继承于 BasicIntQueueDisposable 继承于 AtomicInteger ,通过自身的原子性(自旋/CAS)来消除多线程对 #schedule 的调用。
可以看出 #observeOn 只对下游有影响。
正文到此结束
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
