just
@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> just(T item) {
ObjectHelper.requireNonNull(item, "The item is null");
return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
}
@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> just(T item1, T item2) {
ObjectHelper.requireNonNull(item1, "The first item is null");
ObjectHelper.requireNonNull(item2, "The second item is null");
//多个参数的方法调用了fromArray操作符
return fromArray(item1, item2);
}
just有很多重载方法,我们先来看一个参数的,老套路,直接看ObservableJust,他是Observable的子类
@Override
protected void subscribeActual(Observer<? super T> observer) {
//ScalarDisposable是ObservableScalarXMap类中的静态内部类
ScalarDisposable<T> sd = new ScalarDisposable<T>(observer, value);
observer.onSubscribe(sd);
//ScalarDisposable是一个runnable
sd.run();
}
ScalarDisposable是ObservableScalarXMap类中的静态内部类,本质是一个Runnable,我们重点看run方法
public static final class ScalarDisposable<T>
extends AtomicInteger
implements QueueDisposable<T>, Runnable {
@Override
public void run() {
//这里使用了原子类的方法,重点是调用了onNext和onComplete方法
if (get() == START && compareAndSet(START, ON_NEXT)) {
observer.onNext(value);
if (get() == ON_NEXT) {
lazySet(ON_COMPLETE);
observer.onComplete();
}
}
}
}
fromArray
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
public static <T> Observable<T> fromArray(T... items) {
ObjectHelper.requireNonNull(items, "items is null");
if (items.length == 0) {
return empty();
} else
//这里可以看到,第一个参数的方法调用了just操作符
if (items.length == 1) {
return just(items[0]);
}
//重点看ObservableFromArray
return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));
}
这里直接看ObservableFromArray的subscribeActual方法,我们知道他才是真正的订阅方法
@Override
public void subscribeActual(Observer<? super T> observer) {
FromArrayDisposable<T> d = new FromArrayDisposable<T>(observer, array);
observer.onSubscribe(d);
if (d.fusionMode) {
return;
}
d.run();
}
看FromArrayDisposable的run方法
void run() {
T[] a = array;
int n = a.length;
for (int i = 0; i < n && !isDisposed(); i++) {
T value = a[i];
if (value == null) {
downstream.onError(new NullPointerException("The " + i + "th element is null"));
return;
}
downstream.onNext(value);
}
if (!isDisposed()) {
downstream.onComplete();
}
}
我们看到在for循环中执行了onNext,onComplete,onError方法,也就是说我们出入的数据被依次发送给下游的观察者
map
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
@Override
public void subscribeActual(Observer<? super U> t) {
//重点看MapObserver
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) {
//done在onError 和 onComplete以后才会是true，默认这里是false
if (done) {
return;
}
//默认sourceMode是0
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
//这一步执行变换,将上游传过来的T，利用Function转换成下游需要的v
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
//变换后传递给下游Observer
downstream.onNext(v);
}
@Override
public int requestFusion(int mode) {
return transitiveBoundaryFusion(mode);
}
@Nullable
@Override
public U poll() throws Exception {
T t = qd.poll();
return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
}
}
flatMap
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
return flatMap(mapper, false);
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, boolean delayErrors) {
return flatMap(mapper, delayErrors, Integer.MAX_VALUE);
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, boolean delayErrors, int maxConcurrency) {
return flatMap(mapper, delayErrors, maxConcurrency, bufferSize());
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
if (this instanceof ScalarCallable) {
@SuppressWarnings("unchecked")
T v = ((ScalarCallable<T>)this).call();
if (v == null) {
return empty();
}
return ObservableScalarXMap.scalarXMap(v, mapper);
}
return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize));
}
ObservableFlatMap
@Override
public void subscribeActual(Observer<? super U> t) {
if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
return;
}
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}
MergeObserver
@Override
public void onNext(T t) {
// safeguard against misbehaving sources
if (done) {
return;
}
ObservableSource<? extends U> p;
try {
p = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null ObservableSource");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
upstream.dispose();
onError(e);
return;
}
//maxConcurrency 默认大小是 Integer.MAX_VALUE
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
if (wip == maxConcurrency) {
sources.offer(p);
return;
}
wip++;
}
}
//看这里
subscribeInner(p);
}
void subscribeInner(ObservableSource<? extends U> p) {
for (;;) {
if (p instanceof Callable) {
if (tryEmitScalar(((Callable<? extends U>)p)) && maxConcurrency != Integer.MAX_VALUE) {
boolean empty = false;
synchronized (this) {
p = sources.poll();
if (p == null) {
wip--;
empty = true;
}
}
if (empty) {
drain();
break;
}
} else {
break;
}
} else {
//默认走这里
InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
if (addInner(inner)) {
p.subscribe(inner);
}
break;
}
}
}
InnerObserver
@Override
public void onNext(U t) {
//默认是NONE
if (fusionMode == QueueDisposable.NONE) {
parent.tryEmit(t, this);
} else {
parent.drain();
}
}
void tryEmit(U value, InnerObserver<T, U> inner) {
//用了 AtomicInteger 的同步机制
if (get() == 0 && compareAndSet(0, 1)) {
downstream.onNext(value);
if (decrementAndGet() == 0) {
return;
}
} else {
SimpleQueue<U> q = inner.queue;
if (q == null) {
q = new SpscLinkedArrayQueue<U>(bufferSize);
inner.queue = q;
}
q.offer(value);
if (getAndIncrement() != 0) {
return;
}
}
drainLoop();
}
MergeObserver 继承了 AtomicInteger，所以这里的tryEmit方法就利用了 AtomicInteger 的同步机制，所以同时只会有一个 value 被 actual Observer 发射，由于 AtomicInteger CAS锁只能保证操作的原子性，并不保证锁的获取顺序，是抢占式的，所以最终数据的发射顺序并不是固定的（同一个Observable发出的数据是有序的）
compose
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
return wrap(((ObservableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> wrap(ObservableSource<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
if (source instanceof Observable) {
return RxJavaPlugins.onAssembly((Observable<T>)source);
}
return RxJavaPlugins.onAssembly(new ObservableFromUnsafeSource<T>(source));
}
compose是将一系列操作符打包封装到一起方便复用,compose 区分于lift(lift是map,flatMap等操作符的根本),compose,是对ObservableSource本身进行操作的,上面的apply(this),里面this就是代表ObservableSource自己,而lift是对ObservableSource发送的数据进行操作的.
lift给了我们封装Observer的机会，compose给了我们封装Observable的机会。
onErrorResumeNext
拦截error,可以在onError之前处理throwable
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> onErrorResumeNext(final ObservableSource<? extends T> next) {
ObjectHelper.requireNonNull(next, "next is null");
return onErrorResumeNext(Functions.justFunction(next));
}
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> onErrorResumeNext(Function<? super Throwable, ? extends ObservableSource<? extends T>> resumeFunction) {
ObjectHelper.requireNonNull(resumeFunction, "resumeFunction is null");
return RxJavaPlugins.onAssembly(new ObservableOnErrorNext<T>(this, resumeFunction, false));
}
@Override
public void subscribeActual(Observer<? super T> t) {
OnErrorNextObserver<T> parent = new OnErrorNextObserver<T>(t, nextSupplier, allowFatal);
t.onSubscribe(parent.arbiter);
//真正的订阅
source.subscribe(parent);
}
@Override
public void onError(Throwable t) {
if (once) {
if (done) {
RxJavaPlugins.onError(t);
return;
}
downstream.onError(t);
return;
}
once = true;
if (allowFatal && !(t instanceof Exception)) {
downstream.onError(t);
return;
}
ObservableSource<? extends T> p;
try {
//第一次默认走这里,并没调用下游Observer的onError
p = nextSupplier.apply(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
downstream.onError(new CompositeException(t, e));
return;
}
if (p == null) {
NullPointerException npe = new NullPointerException("Observable is null");
npe.initCause(t);
downstream.onError(npe);
return;
}
//这里实现了订阅
p.subscribe(this);
}
onErrorReturn,onExceptionResumeNext,onErrorReturnItem与onErrorResumeNext都可拦截error
doOnError
@Override
public void onError(Throwable t) {
//默认false
if (done) {
RxJavaPlugins.onError(t);
return;
}
done = true;
try {
onError.accept(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
t = new CompositeException(t, e);
}
//调用了下游Observer的onError
downstream.onError(t);
try {
onAfterTerminate.run();
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
RxJavaPlugins.onError(e);
}
}
doOnError会在onError之前执行,但它并不会拦截异常，该Throwable还是会发送至onError
