先来看一下使用demo
Observable.create(new ObservableOnSubscribe<String>() {
@Override public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("test");
}
}).flatMap(new Function<String, ObservableSource<String>>() {
@Override public ObservableSource<String> apply(String s) throws Exception {
return Observable.just(s + " by flatMap");
}
}).subscribe(new Consumer<String>() {
@Override public void accept(String s) throws Exception {
// accept "test by flatMap"
}
});
先看Observable源码中,我们上面用的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, int maxConcurrency, int bufferSize) {
// 非空判断，忽略
ObjectHelper.requireNonNull(mapper, "mapper is null");
// 这两个都是大于0判断，忽略
ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
// 我们并未实现ScalarCallable接口，所以不会进入 if 中
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));
}
首先看下RxJavaPlugins.onAssembly()方法
/**
* Calls the associated hook function.
* @param <T> the value type
* @param source the hook's input value
* @return the value returned by the hook
*/
@SuppressWarnings({ "rawtypes", "unchecked" })
@NonNull
// 根据注释可以知道这是一个关于hook的方法
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
// 我们使用时默认都没有hook，所以 f 为 null，不会进入 if中
if (f != null) {
return apply(f, source);
}
// 这里直接返回了 source，即 传递过来的 new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize)
return source;
}
那么看下ObservableFlatMap类，先看它的subscribeActual方法，这个方法是当ObservableFlatMap被订阅时真正调用的方法。源码如下：
@Override
public void subscribeActual(Observer<? super U> t) {
// 不会进入，忽略
if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
return;
}
// 使用MergeObserver包装原Observer
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}
继续看MergeObserver的 onNext方法
@Override
public void onNext(T t) {
// safeguard against misbehaving sources
if (done) {
return;
}
ObservableSource<? extends U> p;
try {
// mapper.apply(t) 将调用上面demo中的
// @Override public ObservableSource<String> apply(String s) throws Exception {
// return Observable.just(s + " by flatMap");
// }
// 并将 Observable.just(s + " by flatMap");的返回值赋值给 p
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中
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
if (wip == maxConcurrency) {
sources.offer(p);
return;
}
wip++;
}
}
// 最终会执行这里
subscribeInner(p);
}
再继续往下看subscribeInner源码
@SuppressWarnings("unchecked")
void subscribeInner(ObservableSource<? extends U> p) {
for (;;) {
// p 并不是Callable，所以不会进入 if
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 {
// 又将MergeObserver包装成 InnerObserver
InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
if (addInner(inner)) {
p.subscribe(inner);
}
break;
}
}
}
可以看到又将MergeObserver包装成 InnerObserver ，直接看InnerObserver的onNext方法
@Override
public void onNext(U t) {
// funsionMode 默认是 None，进入 if，调用的是上面的 MergeObserable 的 tryEmit 方法
if (fusionMode == QueueDisposable.NONE) {
parent.tryEmit(t, this);
} else {
parent.drain();
}
}
MergeObserable 的 tryEmit 方法
void tryEmit(U value, InnerObserver<T, U> inner) {
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发出的数据是有序的）
如果没有获取到锁，就会将要发射的数据放入 队列中，drainLoop 方法会循环去获取队列中的 数据，然后发射，由于篇幅有限，更详细的调用过程大家可以看源码。
