前言
Rxjava是NetFlix出品的Java框架， 官方描述為 a library for composing asynchronous and event-based programs using observable sequences for the Java VM，翻譯過來就是“使用可觀察序列組成的一個異步地、基於事件的響應式編程框架”。一個典型的使用示范如下：
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
String s = "1234";
//執行耗時任務
emitter.onNext(s);
}
}).map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return Integer.parseInt(s);
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe();
本文要講的主要內容是Rxjava的核心思路，利用一張圖並結合源碼分析Rxjava的實現原理，至於使用以及其比較深入的內容，比如不常用的操作符，背壓等，讀者可以自行學習。另外提一句，本文采用的Rxjava版本是2.2.3，Rxjava最新版本是3.x.x，感興趣的可以自行閱讀，但相信其最核心的原理是不會變化的。另外，本文篇幅較長，最好的效果是邊看本文邊到源碼中體會，如果讀者沒有耐心讀完，可以只看圖片和頭尾。
正題
先放出本文最重要的圖：
Rxjava的核心思路被總結在了圖中，本文分為兩部分，第一部分講圖中的三條流和事件傳遞，第二部分講線程切換的原理，下面進入正題。
流式構建和事件傳遞
在講之前，先提一點，在Rxjava中，有Observable和Observer這兩個核心的概念，但是它們在發生訂閱時，跟普通的觀察者模式寫法不太一樣，因為常識來講，應該是觀察者去訂閱(subscribe)被觀察者，但是Rxjava為了其基於事件的流式編程，只能反着來，observable去訂閱observer，所以在rxjava中，subscribe可以理解“注入”觀察者。
首先我們看上面的圖片，先簡單解釋一下：圖中方形的框代表的是Observable，因為它代表節點，所以用Ni表示，圓形框代表的是觀察者Observer，用Oi標識，后面加括號的意思是Oi持有其下游Observer的引用，左側代表上游，右側代表下游。圖片里有三條有方向的彩色粗線，代表三個不同的流，這三個流是我們為了分析問題而抽象出來的的，代表從構建到訂閱整個事件的流向，按照時間順序從上到下依次流過，它們的含義分別是：
- 從左往右的構建流：用來構建整個事件序列，這個流表征了整個鏈路的構建過程，相當於構造方法。
- 從右往左的訂閱流：當最終訂閱(subscribe方法)這個行為發生的時候，每個節點從右向左依次執行訂閱行為。
- 從左往右的觀察者回調流：當事件發生以后，會通過這個流依次通知給各個觀察者。
我們依次分析這三條流：
構建流
在使用Rxjava時，其流式構建流程是很大的特色，避免了傳統回調的繁瑣。怎么實現的呢？使用過Rxjava的讀者應該都知道，Rxjava的每一步構建過程api都是相同的，這是因為每一步的函數返回結果都是一個Observable，Observable提供了Rxjava所有的功能。那么Obsevable在Rxjava中到底扮演一個什么角色呢？事實上，其官方定義就已經告訴我們答案了，前言里官方定義中有這樣一段：“using Observable sequences”，所以說，Obsevable就是構建流的組件，我們可以看成一個個節點，這些節點串起來組成整個鏈路。Observable這個類實現了一個接口:ObservableSource,這個接口只有一個方法:subscribe(observer)，也就是說，所有的Obsevable節點都具有訂閱這個功能，這個功能很重要，是訂閱流的關鍵，待會會講。總結一下：
在我們編寫Rxjava代碼時，每一步操作都會生成一個新的Observable節點(沒錯，包括ObserveOn和SubscribeOn線程變換操作)，並將新生成的Observable返回，直到最后一步執行subscribe方法
無論是構建的第一步 create方法，還是observeOn,subscribeOn變換線程方法，還是各種操作符比如map,flatMap等，都會生成對應的Observable，每個Observble中要實現一個最重要的方法就是subscribe，我們看其實現:
public final void subscribe(Observer<? super T> observer) {
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
subscribeActual(observer);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
RxJavaPlugins.onError(e);
throw npe;
}
}
這里提一點，大家看源碼時遇到RxJavaPlugins時直接略過看里面的代碼就好了，它是hook用的，不影響主要流程。所以上面代碼其實只有一行有用：
subscribeActual(observer);
也就是說，每個節點在執行subscribe時，其實就是在調用該節點的subscribeActual方法，這個方法是抽象的，每個節點的實現都不一樣。我們舉個栗子，拿ObseverOn這個操作生成的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> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
//xxx省略
}
其中其父類繼承Observable，所以它是一個Observble。
整個過程有點像builder模式，不同之處是它是生成了新的節點，而builder模式返回的自身。如果你讀過okHttp的源碼，okHttp中攔截器跟這里有些相似，okHttp中會構建多個Chain節點，然后用相應的Intercepter去處理Chain。
我們理解了編寫Rxjava代碼的過程其實就是構建一個一個Observable節點的過程，接下來我們看第二條流。
訂閱流
構建過程只是通過構造函數將一些配置傳給了各個節點，實際還沒有執行任何代碼，只有最后一步才真正的執行訂閱行為。當最后一個節點調用subscribe方法時，是構建流向訂閱流變化的轉折點，我們以圖中為例:最后一個節點是N5，N5節點是最后一個flatmap操作符方法產生的，也就是說，最后是調用這個節點的subscribe方法，這個方法最終也是會調用到subscribeActual方法中去，我們看其源碼：
public final class ObservableFlatMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends ObservableSource<? extends U>> mapper;
final boolean delayErrors;
final int maxConcurrency;
final int bufferSize;
public ObservableFlatMap(ObservableSource<T> source,
Function<? super T, ? extends ObservableSource<? extends U>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
super(source);
this.mapper = mapper;
this.delayErrors = delayErrors;
this.maxConcurrency = maxConcurrency;
this.bufferSize = bufferSize;
}
@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));
}
static final class MergeObserver<T, U> extends AtomicInteger implements Disposable, Observer<T> {
final Observer<? super U> downstream;
final Function<? super T, ? extends ObservableSource<? extends U>> mapper;
}
剛才我們分析了，N5節點是Observable節點，其subscribe方法最后調用的是subscribeActual方法，我們看上面代碼中它的這個方法：前面的判斷語句跳過，第二行：
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
這行代碼需要注意兩點：
- 生成了一個新的Observer，請注意其構造函數中第一個參數t，保存到了downstream這個“下游”變量中，這個t從哪兒傳進來的呢？對於N5節點來說，這個t就是我們代碼中最后一步編寫的Observer，比如我們常用的網絡請求返回后的回調。也就是說，這個新生成的Observer包含了它的“下游”觀察者的引用，在圖片中對應最右邊的圓形框O1(observer)。
- 執行訂閱行為，這里的source是該節點構造函數傳入的source，通過源碼得知其實就是N5節點的上一個節點N4，因此，這里的訂閱行為本質上是讓當前節點的上一個節點訂閱當前節點新生成的Observer。
到這里，我們分析了最后一個節點執行subscribe方法的過程，事實上，每個節點的執行流程都是類似的(subscribeOn節點有些特殊，等會線程調度會將)，也就是說，N5會調用N4的subscribe方法，而在N4的subscribe方法中，又去調用了N3的subscribe....一直到N0會調用source的subscribe方法。總結下來就是：
從最后一個N5節點的訂閱行為開始，依次執行前面各個節點真正的訂閱方法。在每個節點的訂閱方法中，都會生成一個新的Observer，這個Observer會包含“下游”的Observer，這樣當每個節點都執行完訂閱(subscribeActual)后，也就生成了一串Observer，它們通過downstream,upstream引用連接。
以上就是訂閱流的發生過程，簡單講就是下游節點調用上游節點的subscribeActual方法，從而形成了一個調用鏈。
觀察者回調流
當訂閱流執行到最后，也就是第一個節點N0時，我們看發生了什么，首先看看N0節點怎么建立的：
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
生成了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);
source.subscribe(parent);
}
}
所以訂閱流的最終會掉到上面的subscrbeActual方法，它其實還是和其他節點一樣，最主要的還是執行了
source.subscribe(parent)
這行代碼，那么這個節點的source是什么呢？它就是我們事件的源頭啊！
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
String s = "1234";
//執行耗時任務
emitter.onNext(s);
}
})
上面代碼直接拿的開頭的例子，這個source是一個ObservableOnSubscribe，看它的subscribe方法里，這里很重要，這個函數里面其實是訂閱流和觀察者流的轉折點，也就是流在這兒“轉向了”。這里，這個事件源沒有像節點那樣，調用上一個節點的訂閱方法，而是調用了其參數的emitter的onNext方法，這個emitter對應N0節點的什么呢？看代碼知道，時CreateEmitter這個類，我們看這個類里面
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 (!isDisposed()) {
observer.onNext(t);
}
}
//省略
}
看它的onNext方法，執行的是
observer.onNext(t)
observer是誰？構造函數傳進來的，也就是N0節點subscribeActual方法中的observer，這個observer是誰呢？仔細回想一下，前面訂閱流的時候不就是一次訂閱上一個節點生成的Observer嗎，所以這個observer就是前一個節點N1生成的Observer，我們看N1節點，是一個Map，對應的Observable節點里的Observer源碼如下:
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) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
downstream.onNext(v);
//省略后續
名為MapObserver，看它的onNext方法，忽略前面兩個判斷語句，核心就兩句，一個是mapper.apply(t)，另一個就是downstream.onNext(v)。也就是說，這個mapObserver干了兩件事，一個是把上個節點返回的數據進行一次map變換，另一個就是將map后的結果傳遞給下游，下游是什么呢？看了訂閱流的讀者自然知道，就是N2節點的Observer，對應圖中O4,依次類推，我們知道了，事件發生以后，通過各個節點的Observer事件源被層層處理並傳遞給下游，一直到最后一個觀察者執行完畢，整個事件處理完成。
至此，我們三個流分析完畢，接下來，我們開始分析線程調度是怎么實現的。
線程調度
觀察仔細的讀者可能已經看到了，圖中N2節點左側的所有節點和右側的節點顏色不同，我為什么要這樣畫呢？其實里面的玄機就是線程調度，接下來我們分別看subscribeOn和observeOn的線程切換玄機吧。
SubscribeOn
在訂閱流發生的的時候，大多數節點都是直接調用上一個節點的subscribe方法，實現雖有差別，但大同小異。唯一有個最大的不同就是subscribeOn這個節點，我們看源碼：
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> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
普通的節點執行時，大多只是簡單的執行source.subscribe(observer)，但是這個不一樣。先看第二行，它調用了觀察者的onSubscribe方法，熟悉Rxjava的人知道，我們在自定義Observer的時候，里面有這個回調，其發生時機就在此刻。我們接着看最后一行,忽略parent.setDisposable這個邏輯，我們直接看參數里面的東西。
scheduler.scheduleDirect(new SubscribeTask(parent))
看看干了什么：
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}
繼續：
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
w.schedule(task, delay, unit);
return task;
}
創建了一個worker，一個runnable，然后將二者封裝到一個DisposeTask中，最后用worker執行這個task，那么這個worker是什么呢？
@NonNull
public abstract Worker createWorker();
createworker是一個抽象方法，所以需要去找Scheduler的子類，我們回想一下rxjava的使用，如果在子線程中執行，我們一般設置調度器為Schedulers.io(),我們看這個子類的實現:
在IOSchedluer類中:
@Override
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
繼續:
@NonNull
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
Future<?> f;
try {
if (delayTime <= 0) {
f = executor.submit((Callable<Object>)sr);
} else {
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
if (parent != null) {
parent.remove(sr);
}
RxJavaPlugins.onError(ex);
}
return sr;
}
這里的executor就是一個ExecutorService，熟悉線程池的讀者應該知道，這里的submit方法，就是將callable丟到線程池中去執行任務了。
我們回到主線
scheduler.scheduleDirect(new SubscribeTask(parent))
對於io線程的調度器來說，上面的代碼就是將new SubscribeTask(parent)丟到線程池中執行，我們看參數里面的SubscribeTask：
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
source.subscribe(parent);
}
}
看run方法：source.subscribe(parent)，這里的parent跟普通節點一樣，仍然是本節點生成的新的Observer，對於本節點來說，是一個SubscribeOnObserver。因此，我們就知道了，對於subscribeOn這個節點，它跟普通的節點不同之處在於：
SubscribeOn節點在訂閱的時候，將它的上游節點的訂閱行為，以runnable的形式扔給了一個線程池(對於IO調度器來說)，也就是說，當訂閱流流到SubscribeOn節點時，線程發生了切換，之后流向的節點都在切換后的線程中執行。
分析到這里，我們就知道了subscribeOn的線程切換原理了，原來是在訂閱流中塞了一個線程變化操作。我們再看圖中的顏色問題，為什么這個節點上游的節點都是紅色的呢？因為當訂閱流流過這個節點后，后面的節點只是單純的傳遞給上游節點而已，無論是普通的操作符，還是ObserveOn節點，都是簡單的傳遞給上游，沒有做線程切換(注意，ObserveOn是在觀察者流中做的線程切換，待會會講)。
我們再思考一個問題，如果上游還有別的subscribeOn，會發生什么？
我們假設N1節點的map修改程subscribeOn(AndroidScheduler.Main)，也就是說，切換到主線程。我們還是從N2節點開始分析，剛才說到最后會執行到SubscribeTask里的Run方法，注意此時source.subscribe(parent)發生在子線程中，接下來，回調用N1節點的subscribe，N1節點回調用scheduler.scheduleDirect(new SubscribeTask(parent))，方法，此時，因為線程調度器是主線程的，我們看它的代碼：
private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}
看看這個HandlerScheduler的方法:
@Override
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
handler.postDelayed(scheduled, unit.toMillis(delay));
return scheduled;
}
熟悉Android Handler機制的讀者應該很清楚，這里會把N1節點上游的操作，通過Handler機制，扔給主線程操作，雖然這一步是在N2節點的子線程中執行的，但是它之前的事件仍然會在主線程中執行。因此我們有以下結論：
subscribeOn節點影響它前面的節點的線程，如果前面還有多個subscribeOn節點，最終只有第一個，也就是最上游的那個節點生效
接下來我們分析observeOn
ObserveOn
前面的subscribeOn線程切換是在訂閱流中發生的，接下來的ObserveOn比較簡單，它發生在第三條流-觀察者回調流中，我們看ObserveOn節點的源碼:
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
//簡化
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
}
在前面的觀察者流分析時，我們知道，觀察者流是通過onNext()方法傳遞的，我們看最后一行，schedule()，線程切換，所以這個ObserveOn節點其實沒干啥事，就是切換線程了，而且是在onNext回調中切換的。我們進到這個方法：
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
worker是這個節點訂閱時指定的 scheduler.createWorker()， 以主線程觀察為例:
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
if (async) {
message.setAsynchronous(true);
}
handler.sendMessageDelayed(message, unit.toMillis(delay));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
同樣，通過Handler機制，將runnable扔給主線程執行,runnable是誰呢，是this，this就是這個ObserveOnObserver,我們看它的run方法：
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
繼續看drainNormal
void drainNorml() {
//簡化
final Observer<? super T> a = downstream;
T v;
v = q.poll();
a.onNext(v);
}
抓重點，還是把上游的處理結果扔給下游。也就是說observeOn會將它下游的onNext操作扔給它切換的線程中，因此ObserveOn影響的是它的下游，所以我們途中observeOn后面的顏色都是藍的。
同樣我們思考，如果有多個observeOn會發生什么？很簡單，思路同subscribeOn，每個ObserveOn只會影響它下游一直到下一個obseveOn節點的線程，也就是分段的。
總結
到此為止我們就講完了全部內容，包括三條流的原理和線程切換的原理，至於Rxjava的其他功能和原理，限於篇幅，本文不會講解，感興趣的讀者自行閱讀源碼。本文主要為讀者提供了理解Rxjava的思路，真正要去理解它，還是要多看源碼。
在我看來，Rxjava有點像觀察者模式和責任鏈模式的結合，普通的觀察者模式一般是被觀察者通知多個觀察者，而Rxjava則是被觀察者通知第一個Obsever,接下來Observer依次通知其他節點的Observer，形成一個“觀察鏈”，將觀察者模式進行了一種類似鏈式的變換，每個節點又會執行它不同的“職責”，非常巧妙，總結以下就是：
最原始的訂閱事件從最后一個節點開始，沿着Obsevable節點往上游傳遞，事件源頭處理完任務后，通知給最上游的觀察者，然后通知沿着Observer鏈條往下游傳遞，直到最后一個觀察者結束。
后記，關於flatmap
關於flatmap這個操作符，讀者可能會用到，但理解起來又比較難，我們通過本文，其實就很容易從源碼中理解這個操作符的含義。這里我順便給大家解釋一下，還是看圖：
上圖簡化了整個事件流向，我們對事件源進行了flatmap操作，flatmap在訂閱流的時候跟其他的操作符基本一致，但是在觀察者回調流中卻很不一樣，它在回調流中做了以下內容:
flatmap將上游傳過來的數據進行了一次變換，變成了一個Observable，如何變的是由開發者自定義的，比如圖中下面三個豎着的三個Observable節點流，這條流跟上面的四個Observable節點本質上是一樣的。flatmap這個節點的Obsever將上游的數據轉化成了一個新的Observable流，然后執行這條新的流，當這條新的流走完時，會接着原來的觀察者流繼續走下去。也就是說，flatMap這個操作符將一條新的Observable節點流“插入”到原始的觀察者回調流上去了。
那圖中的橘黃色和紫色的虛線是什么意思呢？
實際上它是flatmap的一種特殊情況，當新插入的流的事件源有多個的時候，這是會產生分流，每個流都會執行一遍下游的原始節點。我們拿下面這個例子來看：
String[] mainArmy = {"第一大隊", "第二大隊", "第三大隊"};
Observable.fromArray(mainArmy)
.flatMap(new Function<String, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(String s) throws Exception {
String[] littleArmy = {s + "的第一小隊", s + "的第二小隊", s + "的第三小隊"};
return Observable.fromArray(littleArmy);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String little) throws Exception {
System.out.println(little);
}
});
這個代碼運行結果是
第一大隊的第一小隊
第一大隊的第二小隊
第一大隊的第三小隊
第二大隊的第一小隊
第二大隊的第二小隊
第二大隊的第三小隊
第三大隊的第一小隊
第三大隊的第二小隊
第三大隊的第三小隊
這個例子也許很好的體現了flatmap這個操作符的意義，把list鋪平展開，而且防止了繁瑣的嵌套循環。但是，雖然flatmap很擅長處理這種問題(我不知道這個操作符是不是為了解決這個問題而設計出來的)，但flatmap的功能卻遠不僅如此，它的本質是在合並Obsevable流，可以做很多事情，比如我們網絡請求的“連環請求”，舉個例子，首先通過書本的Id獲取出版商名字，然后拿到出版商名字后獲取出版社信息。
api.getBookPublisherNameById("01102").flatmap(new Function<String, ObservableSource<PublisherInfo>>() {
@Override
public ObservableSource<PublisherInfo> apply(String s) throws Exception {
return api.getPublisherInfoByName(s);
}
}).subscribe(new COnsume<PublisherInfo>() {
@Override
public void accept(PublisherInfo little) throws Exception {
//獲取到出版社信息
}
})
看完這里，flatmap是不是也蠻好理解的～