一 基本使用

Observable建立

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
     @Override
     public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         //ObservableEmitter 發射器的意思，用來發出事件的，它可以發出三種類型的事件，通過呼叫emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分別發出next事件、complete事件和error事件。
 

     }
 });

Observer建立

Observer<Integer> observer = new Observer<Integer>(){
  @Override
  public void onSubscribe(Disposable d) {
  //開關的意思，當呼叫它的dispose()方法時，它就會切斷事件的接受，導致觀察者接受不到事件。這個可以用作當頁面銷燬時，取消網路請求。
  }

  @Override
  public void onNext(Integer integer) {     
  }

  @Override
  public
 
 void onError(Throwable e) {
  }

  @Override
  public void onComplete() {
  }
};

訂閱被觀察者與觀察者

observable.subscribe(observer);

一個observable可能沒有觀察者訂閱，也可以多個觀察者訂閱，當觀察者和被觀察者建立聯絡後，被觀察者可以通過onNext傳送事件，也可以傳送onComplete標誌事件完成，被觀察者傳送onComplete之後，可以繼續傳送，而觀察者接受到onComplete之後，就不會繼續接收事件。

onComplete和onError必須唯一併且互斥。

subscribe有多個方法的過載，帶有一個引數的Consumer方法，表示下游可以只關心onNext方法。

二 執行緒控制

預設情況下，觀察者和被觀察者工作在主執行緒。通常我們的需求是在子執行緒做耗時操作，然後回到主執行緒中來操作UI。此時我們可以用執行緒控制來達到這一目的。

observable
     .subscribeOn(Schedulers.io())  //subscribeOn被觀察者工作的執行緒
     .observeOn(AndroidSchedulers.mainThread()) //observeOn 觀察者工作的執行緒
     .subscribe(observer);

RxJava內建執行緒選項：
1.Schedulers.io() 代表io操作的執行緒, 通常用於網路,讀寫檔案等io密集型的操作，內部用執行緒池維護。
2.Schedulers.computation() 代表CPU計算密集型的操作, 例如需要大量計算的操作
3.Schedulers.newThread() 代表一個常規的新執行緒
4.AndroidSchedulers.mainThread() 代表Android的主執行緒

注意：
1.多次指定被觀察者的執行緒只有第一次指定的有效, 也就是說多次呼叫subscribeOn() 只有第一次的有效, 其餘的會被忽略.
2.多次指定觀察者的執行緒是可以的, 也就是說每呼叫一次observeOn() , 下游的執行緒就會切換一次.

三 操作符

< 1 >.map

map是RxJava中最簡單的一個變換操作符, 它的作用就是對observable傳送的每一個事件應用一個函式, 使得每一個事件都按照指定的函式去變化.

eg:

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> e) throws Exception {
        e.onNext(1);
        e.onNext(2);
        e.onNext(3);
    }
}).map(new Function<Integer, String>() {
    @Override
    public String apply(Integer integer) throws Exception {
        return "this is " + integer;
    }
}).subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Consumer<String>() {
      @Override
      public void accept(String s) throws Exception {
          System.out.println(s);
      }
  });

通過Map可以將Observable傳送的事件轉換為任意的型別。

< 2 >. flatmap

FlatMap將一個傳送事件的Observable變換為多個傳送事件的Observables，然後將它們發射的事件合併後放進一個單獨的Observable裡.

eg:

Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            e.onNext(1);
            e.onNext(2);
            e.onNext(3);
        }
    }).flatMap(new Function<Integer, ObservableSource<String>>() {
        @Override
        public ObservableSource<String> apply(Integer integer) throws Exception {
            List<String> list = new ArrayList<>();
            for(int i = 0; i<3; i++){
                list.add("I am value " + integer);
            }
            return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
        }
    }).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            System.out.println(s);
        }
    });

注意：flatmap並不能保證事件的順序，如果需要保證事件的順序可以使用concatmap，用法和flatmap一樣。

< 3 >.zip

zip通過一個函式將多個Observable傳送的事件結合到一起，然後傳送這些組合到一起的事件. 它按照嚴格的順序應用這個函式。它只發射與發射資料項最少的那個Observable一樣多的資料。

Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
    }
}).subscribeOn(Schedulers.io());

Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
    }
}).subscribeOn(Schedulers.io());

Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
    @Override
    public String apply(Integer integer, String s) throws Exception {
        return "";
    }
}).subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {

    }
});

四. 背壓 Backpressure

< 1 >.原理引入

當observable和observer工作在不同執行緒時，它們之間不能直接通訊，當observable傳送一個事件時，是放在一個容器（具體就是一個佇列）裡面的，當容器裡面有事件時，才會向observer傳送事件。而當observable傳送事件太快，observer取事件太慢，容器就會一直放事件，最後會因為容器事件越來越多，最後導致容器OOM了。

這種observable產生事件的速度很快，而observer處理很慢，事件會堆積起來，最終擠爆你的記憶體，導致程式崩潰（MissingBackpressureException異常）。

至此，就可以引出背壓(Backpressure)的概念：背壓是指在非同步場景中，被觀察者傳送事件速度遠快於觀察者的處理速度的情況下，一種告訴上游的被觀察者降低傳送速度的策略。

關於背壓，可以看這裡

< 2 >.解決方案

a. 使用filter過濾掉不需要的資料
b. 使用sample每隔一段時間從observable取出事件，傳送給下游。
c. 延時降低observable傳送事件的速度。
d.使用flowable解決。

大致就是從數量和速度上解決。

五.Flowable

這裡把被觀察者指定為flowable，觀察者指定為subscriber。

eg:

Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(FlowableEmitter<Integer> e) throws Exception {        
        }
        //流速不均勻直接拋異常
    }, BackpressureStrategy.ERROR);

    Subscriber<Integer> subscriber = new Subscriber<Integer>() {
        @Override
        public void onSubscribe(Subscription s) {         
        }

        @Override
        public void onNext(Integer integer) {
        }

        @Override
        public void onError(Throwable t) {
        }

        @Override
        public void onComplete() {
        }
    };
    flowable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);

和observable不同的是，這裡增加了一個引數，用來處理上下游流速不均衡的時候應該怎麼處理的辦法，在例子中直接丟擲了一個BackpressureStrategy.ERROR異常，也就是MissingBackpressureException異常。同時觀察者中不是再使用Disposable，而是使用Subscription，與Disposable不同的是，這種採用的是響應式拉取，當在onSubscribe方法中呼叫了request方法時，observable會根據request的事件數量發相應數量的事件。observable不會再一直髮事件。如果不呼叫request方法，在同一執行緒，就會拋MissingBackpressureException異常，在不同執行緒，observable就會先把事件放到一個容器，大小為128。當observer呼叫request才會傳送事件。

參考