四部分：事件流源頭(source/observable)，對(duì)事件流的操作(operator/transformer)，對(duì)最終事件流進(jìn)行響應(yīng)(subscriber/observer)，以及整個(gè)過程的調(diào)度(scheduler)。

再看文檔

• The Observable Contract
  • Notifications，observable與observer之間的通信包括：OnNext，OnCompleted，OnError，OnSubscribe；observer與observable之間的通信包括：Subscribe，Unsubscribe，Request；
  • observable可能會(huì)觸發(fā)零次或多次OnNext，每次發(fā)出一個(gè)item，最終可能會(huì)有一次OnCompleted或OnError，這兩者不可能都被觸發(fā)，這兩者之一一旦觸發(fā)，就再也不會(huì)有任何事件了；
  • observable可能不會(huì)觸發(fā)任何OnNext，也可能永遠(yuǎn)不會(huì)終止，也可能既不觸發(fā)任何OnNext，也不終止；
  • observable發(fā)出item必須是串行的，可以是在不同的線程中，但是item之間必須遵循某種順序（串行）；
  • 如果observable并未觸發(fā)OnCompleted/OnError，則observer仍可以和observable通信，即觸發(fā)Unsubscribe/Request；但一旦observable已經(jīng)終止了，observer就不應(yīng)該再對(duì)observable發(fā)出任何指令了；
  • OnError不能傳null，必須提供一個(gè)合適的error；
  • observable終止（釋放資源）之前，必須向observer發(fā)出OnCompleted/OnError通知；
  • observable可能在收到observer的Subscribe通知之后，立即開始OnNext；
  • observer發(fā)出Unsubscribe后，observable會(huì)嘗試停止向observer發(fā)出事件，但是并不保證！
  • 當(dāng)observable發(fā)出OnCompleted/OnError之后，subscription就結(jié)束了，observer無(wú)需再Unsubscribe；
  • 當(dāng)有多個(gè)observer先后subscribe同一個(gè)observable時(shí)，它們接收到的item序列沒有保證，即可能從subscribe之后，始終相同，也可能會(huì)replay，甚至是沒有任何關(guān)系；
  • Backpressure，是可選的，并不是所有的（語(yǔ)言）實(shí)現(xiàn)都包含，即便包含，也可能只是部分實(shí)現(xiàn)；
  • 如果observable實(shí)現(xiàn)了Backpressure，且observer使用backpressure，則observer subscribe時(shí)，observable不會(huì)立即發(fā)出item，相反，它會(huì)向observer發(fā)出一個(gè)OnSubscribe通知；
  • observer收到OnSubscribe通知后，可以向observable發(fā)出Request通知，主動(dòng)請(qǐng)求一定數(shù)量的item；
  • observable收到Request后，正常情況下會(huì)向observer發(fā)出對(duì)應(yīng)數(shù)量的item，但是也可能會(huì)發(fā)出OnCompleted/OnError，甚至在observer發(fā)出Request之前就終止自己；
  • 如果observable不支持backpressure，則收到observer的Request時(shí)，應(yīng)該觸發(fā)OnError，并終止自己；
  • 如果observable產(chǎn)生item的速度超過observer Request的速度，則對(duì)多余的item的處理，取決于observable的策略，或丟棄，或保存排隊(duì)，或者由observer配置策略；
• Hot & Cold Observable
  • observable合適開始發(fā)出item？
  • hot：創(chuàng)建之后立即開始，后續(xù)的observer將只能開到subscribe之后的序列；
  • cold：observer subscribe之后才開始，保證observer看到的是完整的序列；
  • 有的實(shí)現(xiàn)版本（語(yǔ)言）中，還存在Connectable的observable，它的開始時(shí)機(jī)既不是創(chuàng)建時(shí)，也不是被subscribe時(shí)，而是connect方法被調(diào)用時(shí)；
  • 注意：hot, cold, connectable和是否replay是兩個(gè)概念，沒有必然聯(lián)系；此外，創(chuàng)建item的代碼執(zhí)行時(shí)機(jī)也是另外一個(gè)概念，它取決于代碼的求值時(shí)機(jī)；
• Operator，文檔包含的是有哪些operator，以及如何選擇operator，詳見文檔。
• Single
  • RxJava的特殊實(shí)現(xiàn)，只會(huì)發(fā)出一個(gè)item，或者OnError，所以subscribe時(shí)，只需提供onSuccess和onError即可；
  • onSuccess和onError只有其中之一會(huì)被調(diào)用，且只會(huì)被調(diào)用一次；是否一定會(huì)有其中之一被調(diào)用？如果不考慮無(wú)限等待的話，可以說會(huì)；
• Subject
  • Subject是某些Rx實(shí)現(xiàn)版本的內(nèi)容，它既是observable，也是observer
  • 作為observer，它可以subscribe到其他的observable，作為observable，它可以被其他observer subscribe，它所subscribe的observable發(fā)出的item，它將轉(zhuǎn)發(fā)出去，而且它也可以自行發(fā)出item；
  • 如果observable是cold的，當(dāng)subject subsrcibe這個(gè)observable時(shí)，它將開始發(fā)出item，因此用一個(gè)subject去subscribe cold observable，可以達(dá)到把cold變hot的效果（此時(shí)subject作為observable，就是hot的了）；
  • AsyncSubject，只在源observable結(jié)束時(shí)，發(fā)出源observable發(fā)出的最后一個(gè)item，如果源observable沒有發(fā)出任何item，則它也不會(huì)發(fā)出任何item；如果有多個(gè)observer subscribe它，它會(huì)始終發(fā)出同一item；如果源observable發(fā)生錯(cuò)誤，它也將發(fā)出錯(cuò)誤，不會(huì)發(fā)出任何item；
  • BehaviorSubject，它被subscribe時(shí)，將會(huì)重放此前最后一個(gè)item（或者一個(gè)默認(rèn)item）；如果subscribe發(fā)生在onError之后，那observer將只會(huì)收到onError；
  • PublishSubject，不會(huì)重放；hot，創(chuàng)建之后就會(huì)開始發(fā)出item（如果有），所以可能會(huì)有item未被任何observer接收到；如果源observable已經(jīng)終止（onCompleted/onError），則它對(duì)后續(xù)的observer只會(huì)發(fā)出相應(yīng)的終止事件；
  • ReplaySubject，會(huì)重放所有的item；有的版本中，當(dāng)歷史item過多，或者時(shí)間過長(zhǎng)，它會(huì)丟棄老的item；使用ReplaySubject時(shí)，不能在多線程調(diào)用onNext/onCompleted/onError方法，否則會(huì)導(dǎo)致非串行，有悖于observable contract，導(dǎo)致無(wú)法確定重放順序；
  • [RxJava] SerializedSubject，可以把普通的subject轉(zhuǎn)化為可以多線程調(diào)用onNext/onCompleted/onError的subject；
  • [RxJava] TestSubject，TestScheduler，用于測(cè)試、調(diào)試；
• Relay
  • 使用Subject可以把已有非Rx API適配為Rx API，但是一旦subject終止之后，就必須重新創(chuàng)建一個(gè)subject才能繼續(xù)接受item了；
  • Relay和Subject類似，但是不會(huì)終止，它繼承自O(shè)bservable，實(shí)現(xiàn)了Action1（而不是Observer）；

原理

• subscribe原理，引用自給 Android 開發(fā)者的 RxJava 詳解

http://wiki.jikexueyuan.com/project/notes/images/rx_subscribe.png" alt="rx_subscribe.png" />

注意，選中的部分，應(yīng)該是subscribe()而不是subscriber()。

• lift變換原理，引用自給 Android 開發(fā)者的 RxJava 詳解

http://wiki.jikexueyuan.com/project/notes/images/rx_lift.png" alt="rx_lift.png" /> http://wiki.jikexueyuan.com/project/notes/images/rx_lift_2.png" alt="rx_lift_2.png" />

• subscribeOn和observeOn原理：也是用lift實(shí)現(xiàn)，通過相應(yīng)Operator實(shí)現(xiàn)線程的切換

細(xì)節(jié)

• just, from等操作均是在創(chuàng)建時(shí)執(zhí)行，而非subscribe時(shí)，很顯然，因?yàn)閖ava函數(shù)調(diào)用傳遞的是值，所以會(huì)先eval；create, defer等操作均是在subscribe時(shí)執(zhí)行；create多次subscribe只會(huì)執(zhí)行一次，defer多次subscribe會(huì)執(zhí)行多次（create、defer，call函數(shù)內(nèi)的代碼每次subscribe均會(huì)被執(zhí)行）；ref
• hot v.s. cold Observable
  • cold：當(dāng)Observable被subscribe時(shí)才開始發(fā)射item；（后來的subscriber同樣會(huì)收到其subscribe之前發(fā)射的item；retrofit實(shí)現(xiàn)的是cold；每次subscribe時(shí)，發(fā)射item的代碼都會(huì)被執(zhí)行）
  • hot：創(chuàng)建之后就會(huì)開始發(fā)射item，不管是否被subscribe；（后來的subscriber不會(huì)收到其subscribe之前發(fā)射的item；）
• map v.s. flatMap
• 將cold observable通過cache轉(zhuǎn)換成hot之后，再在別處subscribe他們，有其應(yīng)用場(chǎng)景：第一次subscribe時(shí)并不關(guān)心結(jié)果，但是后面某處想要獲取結(jié)果，又不希望再次執(zhí)行創(chuàng)建observable的過程；類似于預(yù)取思想；
• 怎么感覺有問題。。。
• concat v.s. merge：concat不會(huì)讓參數(shù)observable發(fā)射的item之間重疊，而merge可能會(huì)；concat傳入的參數(shù)順序是有影響的，merge沒影響；
• share v.s. replay
  • replay
    • 需要調(diào)用connect方法后，observable才開始發(fā)射；
    • 后subscribe的subscriber會(huì)在subscribe的瞬間先收到其subscribe之前發(fā)射的所有item，然后再正常收到剩下的item；
    • subscriber unsubscribe后將不會(huì)再接收到item，但也不會(huì)有onCompleted事件，且對(duì)其他subscriber不影響；
  • 實(shí)現(xiàn)的功能都需要測(cè)試驗(yàn)證，不能憑經(jīng)驗(yàn)、也不能看博客，也不能僅看文檔；

應(yīng)用場(chǎng)景

• Scheduler線程切換
• Retrofit結(jié)合RxJava做網(wǎng)絡(luò)請(qǐng)求框架
• RxJava代替EventBus進(jìn)行數(shù)據(jù)傳遞：RxBus
• 解決嵌套回調(diào)（callback hell）問題
• RxJava進(jìn)行數(shù)組、list的遍歷
• 使用debounce做textSearch
• 使用interval做周期性操作。當(dāng)有“每隔xx秒后執(zhí)行yy操作”類似的需求的時(shí)候，想到使用interval
• 使用timer做定時(shí)操作。當(dāng)有“x秒后執(zhí)行y操作”類似的需求的時(shí)候，想到使用timer
• 使用merge合并兩個(gè)數(shù)據(jù)源
• 使用combineLatest合并最近N個(gè)結(jié)點(diǎn)
  • 還存在另一個(gè)類似的operator：withLatestFrom，他們的使用場(chǎng)景區(qū)分如下（摘自）：
  • If I want to combine Observables that seem completely independent to each other, then I need combineLatest.
  • If I want to just map a certain Observable while using information from other Observables, then I need withLatestFrom.
• 使用concat和first做緩存
• 使用throttleFirst防止按鈕重復(fù)點(diǎn)擊
• 使用schedulePeriodically做輪詢請(qǐng)求
• 使用timeInterval，獲取本次事件與上次事件之間的時(shí)間間隔
• 響應(yīng)式的界面

Code review

• part I