四部分：事件流源頭(source/observable)，對事件流的操作(operator/transformer)，對最終事件流進行響應(subscriber/observer)，以及整個過程的調度(scheduler)。

再看文檔

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

原理

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

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

注意，選中的部分，應該是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實現(xiàn)，通過相應Operator實現(xiàn)線程的切換

細節(jié)

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

應用場景

• Scheduler線程切換
• Retrofit結合RxJava做網絡請求框架
• RxJava代替EventBus進行數(shù)據(jù)傳遞：RxBus
• 解決嵌套回調（callback hell）問題
• RxJava進行數(shù)組、list的遍歷
• 使用debounce做textSearch
• 使用interval做周期性操作。當有“每隔xx秒后執(zhí)行yy操作”類似的需求的時候，想到使用interval
• 使用timer做定時操作。當有“x秒后執(zhí)行y操作”類似的需求的時候，想到使用timer
• 使用merge合并兩個數(shù)據(jù)源
• 使用combineLatest合并最近N個結點
  • 還存在另一個類似的operator：withLatestFrom，他們的使用場景區(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防止按鈕重復點擊
• 使用schedulePeriodically做輪詢請求
• 使用timeInterval，獲取本次事件與上次事件之間的時間間隔
• 響應式的界面

Code review

• part I