RxJava有4個(gè)角色Observable柄沮、Observer、Subscriber和Suject盟广，Observable和 Observer 通過subscribe方法實(shí)現(xiàn)訂閱關(guān)系闷串，Observable就可以在需要的時(shí)候通知Observer。

RxJava的使用

1筋量，創(chuàng)建Observer（觀察者）

它決定事件觸發(fā)的時(shí)候有怎樣的行為烹吵。

Subscriber subscriber = new Subscriber<String>(){
    @Override
    public void onCompleted(){//事件隊(duì)列完結(jié)，當(dāng)不再有新的onNext觸發(fā)時(shí)調(diào)用作為完成標(biāo)志
    }
    @Override
    public void onError(){//在事件處理過程中出現(xiàn)異常時(shí)會觸發(fā)同時(shí)隊(duì)列終止
    }
    @Override
    public void onNext(){//將要處理的事件添加到事件隊(duì)列
    }
    @Override
    public void onstart(){//在事件還未發(fā)送前調(diào)用可以做一些準(zhǔn)備工作
    }
}

Observer是一個(gè)接口桨武，Subscriber是在Observer上進(jìn)行了擴(kuò)展肋拔，也可以使用Observer創(chuàng)建觀察者

Observer<String> observer = new Observer<String>(){
    @Override
    public void onCompleted(){}
    @Override
    public void onError(){}
    @Override
    public void onNext(){}
}

2，創(chuàng)建Observable（被觀察者）

它決定了什么時(shí)候觸發(fā)事件以及觸發(fā)怎樣的事件

Observable observable = Observable.create(new Observable.OnSubscribe<String>(){
    @Override
    public void call(Subscriber<? super String> subscriber){
    subscriber.onNext("123")//調(diào)用方法將事件添加到隊(duì)列
    subscriber.onNext("456")
    subscriber.onCompleted()
    }
})

//簡化寫法利用just和from實(shí)現(xiàn)

Observable observable = Observable.just("123","456")//依次調(diào)用onNext方法和onCompleted方法呀酸。
String[] words = {"123","456"}
Observable observable = Observable.from(words)//依次調(diào)用onNext方法和onCompleted方法凉蜂。

3，Subscribe（訂閱）

將觀察者和被觀察者進(jìn)行關(guān)聯(lián)observable.subscribe(subscriber)

RxJava的Subject

Subject 既可以是一個(gè) Observer 也可以是一個(gè) Observerable性誉，它是連接 Observer 和Observerable的橋梁窿吩。 因此，Subject可以被理解為Subject=Observable+Observer
1错览，PublishSubject：只會把在訂閱發(fā)生的時(shí)間點(diǎn)之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者纫雁，因此為了防止數(shù)據(jù)丟失可以在所有觀察者都訂閱完成后在發(fā)送數(shù)據(jù)。
2倾哺，BehaviorSubject：當(dāng)Observer訂閱BehaviorSubject時(shí)轧邪，它開始發(fā)射原始Observable最近發(fā)射的數(shù)據(jù)。如果此時(shí)還沒有收到 任何數(shù)據(jù)羞海，它會發(fā)射一個(gè)默認(rèn)值忌愚，然后繼續(xù)發(fā)射其他任何來自原始Observable的數(shù)據(jù)。如果原始的 Observable因?yàn)榘l(fā)生了一個(gè)錯(cuò)誤而終止扣猫，BehaviorSubject將不會發(fā)射任何數(shù)據(jù)菜循，但是會向Observer傳遞一個(gè) 異常通知。
3，ReplySubject：不管Observer何時(shí)訂閱ReplaySubject癌幕，ReplaySubject均會發(fā)射所有來自原始Observable的數(shù)據(jù)給 Observer衙耕。不同類型的ReplaySubject用于限定Replay的范圍，例如設(shè)定Buffer的具體大小勺远，或者設(shè)定具體的時(shí)間范圍橙喘。如果使用ReplaySubject作為Observer，注意不要在多個(gè)線程中調(diào)用onNext胶逢、onComplete 和onError方法厅瞎。這可能會導(dǎo)致順序錯(cuò)亂，并且違反了Observer規(guī)則初坠。
4和簸，AsyncSubject：當(dāng)Observable完成時(shí)，AsyncSubject只會發(fā)射來自原始Observable的最后一個(gè)數(shù)據(jù)碟刺。如果原始的 Observable 因?yàn)榘l(fā)生了錯(cuò)誤而終止锁保，AsyncSubject 將不會發(fā)射任何數(shù)據(jù)，但是會向Observer傳遞一個(gè)異常通知半沽。

RxJava操作符

包括defer爽柒、range、interval者填、start浩村、repeat、timer

創(chuàng)建操作符

1占哟，interval
創(chuàng)建一個(gè)按固定時(shí)間間隔發(fā)送整數(shù)序列的Observable心墅，相當(dāng)于定時(shí)器

Observable.interval(6,TimeUnit.SECONDS)//間隔6秒發(fā)送
                  .subscribe(new Action1<Long>(){
                  @Override
                  public void call(Long mLong){
                      //TODO
                      } 
                  })

2，range
創(chuàng)建發(fā)射指定范圍整數(shù)序列的Observable重挑，可以用于代替for循環(huán)嗓化，第一個(gè)參數(shù)為起始值且不小于0棠涮，第二個(gè)參數(shù)為終值谬哀，左閉右開。

Observable.range(0,8)
                  .subscribe((new Action1<Interger>(){
                  @Override
                  public void call(Integer integer){
                      //TODO
                      } 
                  })

3严肪，repeat
創(chuàng)建一個(gè)N次重復(fù)發(fā)射特定數(shù)據(jù)的Observable

Observable.range(0,8)
                   .repeat(3)//重復(fù)執(zhí)行三次0-7循環(huán)
                  .subscribe((new Action1<Integer>(){
                  @Override
                  public void call(Integer integer){
                      //TODO
                      } 
                  })

變換操作符

包括map史煎、flatMap、cast驳糯、concatMap篇梭、flatMapIterable、buffer酝枢、groupBy
1恬偷，map
通過指定一個(gè)Func對象，將Obserable轉(zhuǎn)換為一個(gè)新的Observable對象并發(fā)射帘睦，觀察者將接收到新的Observable處理袍患。例如利用map來處理域名更換：

final String Host = "http://baidu.com/"
Observable.just("image").map(new Func1<String,String>(){
    @Override
    public String call(String s){
      return Host+s
    }
}).subscribe(new Action1<String>(){
                  @Override
                  public void call(String s){
                      //TODO
                      } 
                  })

2坦康，flatMap、cast
flatMap操作符將Observable發(fā)射的數(shù)據(jù)集合變換為Observable集合诡延，然后將這些Observable發(fā)射的數(shù)據(jù)平坦的放入一個(gè)單獨(dú)的Observable滞欠，cast操作符的作用是強(qiáng)制將Observable發(fā)射的所有數(shù)據(jù)轉(zhuǎn)換為指定的類型。例如在多個(gè)請求接口前添加host：

final String Host="http://baidu.com"
Lsit<String> list = new ArrayList<>()
list.add("image1")
list.add("image2")
list.add("image3")
//利用flatMap將list轉(zhuǎn)換為Observable集合并再放入一個(gè)單獨(dú)的Observable中發(fā)射肆良，交叉執(zhí)行不保證發(fā)射順序筛璧。
Observable.from(list).flatMap(new Func1<String,Observable<?>>(){
    @Override
    public Observable<?> call(String s){
    return Observable.just(Host+s)
    }
//cast將Observable的數(shù)據(jù)轉(zhuǎn)換為String類型
}).cast(String.class).subscribe(new Action1<String>(){
                  @Override
                  public void call(String s){
                      //TODO
                      } 
                  })

3，concatMap
concatMap與flatMap操作符一致惹恃，解決了flatMap的交叉問題夭谤，提供了一種能夠把發(fā)射值連續(xù)在一起的函數(shù)，而不合并他們巫糙。
4沮翔，flatMapIterable
可以將數(shù)據(jù)包裝成Iterable，我們在Iterable中對數(shù)據(jù)進(jìn)行處理曲秉。

Observable.just(1,2,3).flatMapIterable(new Func1<Integer,Iterable<Integer>>(){
    @Override
    publc Iterable<Integer> call(Integer s){
    List<Integer> mList = new ArrayList<Integer>()
    mList.add(s+1)//對每個(gè)數(shù)都進(jìn)行+1采蚀，輸出為2，3承二，4
    return mList
    }
}).subscribe(new Action1<Integer>(){
                  @Override
                  public void call(Integer integer){
                      //TODO
                      } 
                  }

5榆鼠，buffer
將原Observable變換為一個(gè)新的Observable，新的Observable每次發(fā)射一組列表值而不是一個(gè)一個(gè)發(fā)射亥鸠。和其類似的有window操作符妆够，window操作符發(fā)射的是Observable而不是數(shù)據(jù)列表

Observable.just(1,2,3,4,5,6)
                  .buffer(3)//緩存容量為3,輸出為兩組每組3個(gè)數(shù)分別輸出
                  .subscribe(new Action1<List<Integer>>(){
                  @Override
                  public void call(List<Interger> integers){
                      //TODO
                      } 
                  })

6，groupBy
用于元素分組负蚊，將原Observable變換為一個(gè)發(fā)射Observable的新的分組后的Observable神妹，每一個(gè)新的Observable都是發(fā)射一組指定數(shù)據(jù)。

Observable<GroupedObservable<String,Object>> groupObservable 
= Observable.just(obj1,obj2,obj3)
                     .groupBy(new Func1<Object,String>){
                     @Override
                     public String call(Object obj){
                      return obj.value//用于確定分組的參數(shù)
                     }
                     }
//對分組后的數(shù)據(jù)輸出
Observable.concat(groupObservable).subscribe(new Action1<Object>(){
                  @Override
                  public void call(Object obj){
                      //TODO
                      } 
                  })

過濾操作符

包括filter家妆、elementAt鸵荠、distinct、skip伤极、take蛹找、skipLast、takeLast哨坪、ignoreElements庸疾、throttleFirst、sample当编、debounce届慈、throttleWithTimeout

組合操作符

包括startWith、merge、concat金顿、zip词渤、combineLastest、join串绩、switch

輔助操作符

包括delay缺虐、DO、subscribeOn礁凡、observeOn高氮、timeout、materialize顷牌、dematerialize剪芍、timeInterval、timestamp窟蓝、to

錯(cuò)誤處理操作符

包括catch罪裹、retry

布爾操作符

包括all、contains运挫、isEmpty状共、exists、sequenceEqual

條件操作符

包括amb谁帕、defaultIfEmpty峡继、skipUntil、skipWhile匈挖、takeUnit碾牌、takeWhile

轉(zhuǎn)換操作符

包括toList、toSortedList儡循、toMap舶吗、toMultiMap、getIterator择膝、nest

RxJava線程控制

如果不設(shè)置線程誓琼，默認(rèn)為在subscribe方法的線程上進(jìn)行回調(diào)，設(shè)置線程需要用到Scheduler
Scheduler.immediate():在當(dāng)前線程運(yùn)行
Scheduler.newThread()：總是啟用新線程
Scheduler.io()：I/O操作使用的Scheduler调榄，和newThread類似踊赠，區(qū)別在于io內(nèi)部實(shí)現(xiàn)的是一個(gè)無數(shù)量上限的線程池，可以重用空閑線程比newThread更有效率每庆。
Scheduler.computation()：計(jì)算使用的模式，使用固定線程池大小為cpu核數(shù)今穿，不要進(jìn)行I/O操作其等待時(shí)間會浪費(fèi)cpu資源缤灵。
Scheduler.trampoline()：在當(dāng)前線程非立即執(zhí)行使用，可以將任務(wù)加入隊(duì)列然后按順序運(yùn)行。
RxAndroid提供的常用的Scheduler
AndroidSchedulers.mainThread()：指定操作在主線程運(yùn)行腮出。
在RxJava中用subscribeOn和observeOn操作符來控制線程帖鸦。

源碼解析

1，RxJava訂閱過程胚嘲。
查看一段RxJava的基本使用代碼作儿。
Observable.create(new Observable.OnSubscribe<Integer>)...
.subscribe(new Subscriber<Integer>)...
查看create方法的定義

public static <T> Observable<T> create(OnSubscribe<T> f){
    return new Observable<T>(hook.onCreate(f))
}

可以看出在create方法中創(chuàng)建了Observable對象并返回，hook表示的是RxJavaObservableExecutionHook馋劈。查看他的onCreate方法定義

public <T> OnSubscribe<T> onCreate(OnSubscribe<T> f){
  return f
}

在RxJavaObservableExecutionHook的onCreate方法中只是返回了傳入得被觀察者對象攻锰。
查看Observable的構(gòu)造方法

protected Observable(OnSubscribe<T> f){
  this.onSubscribe = f//將前邊構(gòu)建的Observable對象賦值給了onSubscribe
}

之后調(diào)用subscribe方法完成訂閱，查看Observable的subscrbie方法

static <T> Subscription subscribe(Subscriber<? super T> subscriber,Observable<T> observble)
...
subscriber.onStart()
if(!(subscriber instancefo SafeSubscriber)){//進(jìn)行類型檢查妓雾，如果不是進(jìn)行封裝
//SafeSubscriber繼承自Subscriber娶吞，在 onCompleted和onError方法調(diào)用時(shí)不會再調(diào)用onNext，且保證onCompleted和onError方法只有一個(gè)執(zhí)行
    subscriber = new SafeSubscriber<T>(subscriber)
}
try{
    hook.onSubscribeStrt(observable, observable.onSubscribe).call(subscriber)
    return hook.onSubscribeReturn(subscriber)
}catch(Throwable e){
    ...
    return Subscriptions.unsubscribed()
}

查看hook的onSubscribeStart方法可以發(fā)現(xiàn)是調(diào)用OnSubscribe.call（subscriber）來完成訂閱的

public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observable,final OnSubscribe<T> onSubscribe){
    return onSubscribe
}

2械姻，RxJava的變換過程
前邊說過map操作符會將源Observable轉(zhuǎn)換為一個(gè)新的Observable妒蛇，查看map方法

public final <R> Observable<R> map(Func1<? super T,? extends R>func){
//OperatorMap實(shí)現(xiàn)了Operator接口的call方法，且在call方法中創(chuàng)建了MapSubscriber并返回
    return lift(new OperatorMap<T,R>(func))
}

lift 方法返回一個(gè)新建的 Observable 對象ob2楷拳，并傳入了一個(gè) OnSubscribeLift對象記為on2,他的構(gòu)造函數(shù)中需要兩個(gè)參數(shù)為onSubscribe 和 operator（OperatorMap）绣夺，在OnSubscribeLift構(gòu)造方法中會拿到開始creat創(chuàng)建的Observable（前邊提過的onSubscribe變量on1），在其call方法中調(diào)用hook.onLift(operator).call(o)方法即調(diào)用的是用OperatorMap的call方法返回MapSubscriber記為sub2欢揖，繼續(xù)執(zhí)行可理解為on1.call(sub2)完成訂閱乐导。在map方法后調(diào)用的subscribe方法傳入Subscriber類型參數(shù)標(biāo)記為sub1。
subscribe方法前面講過浸颓，它會調(diào)用：hook.onSubscribeStart(observable,observable.onSubscribe).call(subscriber) 因?yàn)榇饲罢{(diào)用過map操作符物臂，所以這里傳入的observable.onSubscribe指的是on2。分析RxJava的訂閱過程产上，onSubscribeStart方法會返回onSubscribe棵磷，也就是on2，相當(dāng)于on2.call（sub1）晋涣，on2指的是 OnSubscribeLift仪媒，在on1的call中會調(diào)用sub2的onNext查看MapSubscriber的onNext方法中調(diào)用了actual.onNext(result)方法，actual指的是sub1從而完成了變換過程谢鹊。

map轉(zhuǎn)換圖

3算吩，RxJava的線程切換過程
線程切換主要用到了subscribeOn和observeOn兩個(gè)方法，一個(gè)決定了被觀察者執(zhí)行線程佃扼，一個(gè)決定了觀察者運(yùn)行線程偎巢。

subscribeOn方法定義：

public final Observable<T> subscribeOn(Scheduler shceduler){
  if(this instanceof ScalarSynchronousObservable){
    return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler)
  }
  return create(new OperatorSubscribeOn<T>(this,scheduler))
}

上面代碼create 方法仍舊是生成一個(gè)新的 Observable，并傳入一個(gè)OperatorSubscribeOn 類兼耀。 OperatorSubscribeOn 需要傳入兩個(gè)參數(shù)：第一個(gè)參數(shù) this压昼，指的是我們最先創(chuàng)建的Observable求冷，第二個(gè)參數(shù)是一個(gè)Scheduler。在OperatorSubcribeOn的call方法中調(diào)用Scheduler的createWorker方法會創(chuàng)建Worker然后調(diào)用它的schedule方法窍霞，Worker是線程處理的代理執(zhí)行者匠题。
選取查看Schedulers.newThread()代碼：

poublic static Scheduler newThread(){
  return getInstance().newThreadSchedulere
}

Scheduler 是一個(gè)單例類，其返回自身的 newThreadScheduler 屬性但金。這個(gè)屬性最終指的是 NewThreadScheduler韭山。

public final class NewThreadScheduler extends Scheduler{
  private final ThreadFactory threadFactory
  public NewThreadScheduler(ThreadFactory threadFactory){
    this.threadFactory = threadFactory
  }
  @Override
   public Worker createeWorker(){
      reurn new NewThreadWorker(threadFactory)
   } 
}

此前在 OperatorSubscribeOn 中調(diào)用了 Scheduler 的 createWorker 方法，其實(shí)就是調(diào)用 NewThreadScheduler 的createWorker方法冷溃。NewThreadWorker中使用了ScheduledThreadPool線程池钱磅。OperatorSubscribeOn中調(diào)用了 Worker 的 schedule 方法，而 Worker 指的是NewThreadWorker

public Subscription schedule(final Action0 action,long delayTime,TimeUnit unit){
  if(isUnsubscribed){
    return Subscriptions.unsubscribbed()
  }
  return scheduleActual(action,delayTime,unit)
}
//scheduleActual方法
public ScheduledAction scheduleActual(final Action0 action,long delayTime,TimeUnit unit){
  Action0 decoratedAction = schedulersHook.onSchedule(action)
  ScheduledAction run = new ScheduledAction(decoratedAction)
  Future<?>f
  if(delayTime<=0){
     f=executor.submit(run)
  }else{
    f = executor.schedule(run,delayTime,unit)
  }
  run.add(f)
  return run
}

可以看到最終線程切換的處理均由線程池處理秃诵。

observeOn方法定義

public final Observable<T> observeOn(Scheduler scheduler,boolean delayError,int bufferSize){
  if(this instanceof ScalarSynchronousObservable){  
    return ((ScalarSynchronousObservable<T>this).scalarScheduleOn(scheduler)
  }
  return lift(new OperatorObserveOn<T>(scheduler,delayError,bufferSize)
}

OperatorObserveOn和此前講過的OperatorMap類似在其call方法中創(chuàng)建了ObserveOnSubscriber

ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler,child,delayError,bufferSize)
parent.init()
return parent

ObserveOnSubscriber是Subscriber的子類续搀，在其onNext，onCompleted菠净，onError方法中都調(diào)用了schedule方法禁舷。

protected void schedule(){
    if(counter.getAndIncrement() == 0){
        recursiveScheduler.schedule(this)
    }
}

recursiveScheduler是一個(gè)Worker，this指的是ObserveOnSubscriber毅往，這意味著ObserveOnSubscriber的 onNext方法都被切換到recursiveScheduler的線程做處理牵咙，從而達(dá)到線程切換的目的。