RxJava介紹和使用

最近開始研究RxJava，真坑爹，這個東西不好搞，可能本人的智商堪憂！終于上手了，自己總結一下。

RxJava 最早是後端用的，但是RxJava在後端上倒是不太流行，反而在Android上流行起來了！Android中UI開發需要異步操作，使用RxJava可以随意切換線程，用起來比較友善，代碼比較清晰。

RxJava 類似 觀察者模式。有一個Observable(被觀察者，事件的發起者)和一個觀察者(Subscriber，事件的接收者)。Observable發出一系列的事件，然後Subscriber處理這些事件。

Observable 可以發送零到多個事件;Subscriber會接收它發的每一個事件。如果Observable沒有對應的Subscriber,不會發送任何事件。

如何發送資料

首先我們看一下Observable 是如何建立和發送事件的。

1. 我們需要一個Observable，用來發送事件。我們可以通過

   create()

   方法來建立一個Observable。它決定什麼時候觸發事件以及怎麼發送事件。在

   create()

   裡傳入了一個

   OnSubscribe

   ，它會有一個

   call

   方法,它提供了一個

   Subscriber

   來通過

   onNext

   發送事件。發送完畢之後

   Subscriber

   會提供一個

   onComplete

   方法，來結束事件的發送。

   Observable<String> mObservable = Observable.create(new Observable.OnSubscribe<String>() {
       @Override
       public void call(Subscriber<? super String> subscriber) {
           LogUtils.e("開始發送事件");
           subscriber.onNext("Hello World");//通過onNext 發送事件
           subscriber.onCompleted();//結束發送事件
       }
   });
              

2. 然後我還需要一個

   Subscribe

   用來接收接收

   Observable

   發送的事件。

   Subscriber<String> mSubscriber = new Subscriber<String>() {
       //在監聽事件發送剛開始也就是subscribe()剛開始，而事件還未發送之前被調用，可以用于做初始化工作
       @Override
       public void onStart() {
           super.onStart();
           LogUtils.e("開始接收資料");
       }
   
       @Override
       public void onCompleted() {
           LogUtils.e("接收事件結束");
       }
       @Override
       public void onError(Throwable throwable) {
           LogUtils.e("接收事件發生錯誤:" + throwable.getMessage());
       }
       //對應Observable中的onNext對應，Observable中的onNext調用一次，這裡對應也會調用一次
       @Override
       public void onNext(String s) {
           Toast.makeText(MainActivity.this, "" + s, Toast.LENGTH_SHORT).show();
           LogUtils.e(s);
       }
   };
              

3.前面說，如果Observable沒有對應的Subscriber,不會發送任何事件。是以我們還需要

Subscriber

對

Observable

的訂閱。

mObservable.subscribe(mSubscriber);
           

這樣就完成了事件的發送于接收。

建立Observable的方式

在RxJava中建立

Observable

還有其它的方式

1. 如果傳遞的資料是多個同類型的資料，可以使用以下的方式：

   //已發送 “Hello”,”World”，”RxJava”，相當于調用三次onNext,并調用onComplete

   Observable observable = Observable.just(“Hello”, “World”,”RxJava”);

2. 如果傳遞的資料是數組：

   String[] strArr = {"Hello", "World","RxJava"};
   Observable observable = Observable.from(strArr);
              

3. 如果傳遞的資料是集合：

   List<String> strList = new ArrayList<>();
   strList.add("Hello");
   strList.add("World");
   strList.add("RxJava");
   Observable observable = Observable.from(strList);
              

常用的操作符

map

map()在RxJava裡非常常用,它可以對事件進行變換。非常喜歡這個操作符，它可以将事件序列中的對象或整個發送的整個序列進行變換，進行不同的處理。例如：我們想需要傳遞一個圖檔的Bitmap集合，但是我們隻有一個圖檔的URL集合。我們就可以使用map對URL集合進行處理。

List<String> strList = new ArrayList<>();
    strList.add("http://h.hiphotos.baidu.com/zhidao/pic/item/3812b31bb051f81991b9d8dbdcb44aed2f73e787.jpg");
    strList.add("http://cdn.duitang.com/uploads/item/201412/19/20141219154459_3P4G2.jpeg");
    strList.add("http://ww1.sinaimg.cn/crop.7.22.1192.1192.1024/5c6defebjw8epti0r9noaj20xc0y1n0x.jpg");
    Observable observable = Observable.from(strList)
            //map的參數是一個Func1類,它的第一個泛型是指傳遞資料類型，第二個參數是經過轉換過傳回的資料類型
            //這裡我們轉換之前的資料類型是String，處理後傳回的是Bitmap
            .map(new Func1<String, Bitmap>() {
                @Override
                public Bitmap call(String s) {
                     Bitmap bitmap = parseString2Bitmap(s);
                    return bitmap;
                }
            });
           

flatMap

flatMap的功能比較難解釋，我們先舉個例子啊；假如我們有若幹個使用者買了我們若幹的産品，我們想知道哪些産品比較受歡迎，需要我們列印一下這些使用者購買的産品。我們這裡隻列印産品的名稱。

我們先準備一下資料：

List<User> users = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        User user = new User();
        user.id = i;
        user.name = "UserName = " + i;
        user.address = "火星" + i + "号";
        for (int j = 0; j < 5; j++) {
            Product product = new Product();
            product.name = "ProductName" + j;
            product.price = j * j + "";
            product.number = j * 1000 + "";
            user.mProducts.add(product);
        }
        users.add(user);
    }
           

接下來我們來過濾這些使用者：

Observable observable = Observable.from(users)
            //利用flatMap,将之前我們User集合的每個user中的Product集合提取出來建構一個新的Observable,然後再将product一次往下傳遞
            .flatMap(new Func1<User, Observable<Product>>() {
                @Override
                public Observable<Product> call(User user) {
                    //我們在建立第一個Observable時使用User集合，我們把每個User提取出來，擷取其中的Product集合，并利用Product集合建立一個新的Observable
                    return Observable.from(user.mProducts);
                }
                //通過上面的flatMap,使用Product集合建立新的Observable，我們在通過map将product的名稱提取出來，并且傳遞到Subscribe
            }).map(new Func1<Product, String>() {
                @Override
                public String call(Product p) {
                    return p.name;
                }
            })
           

這裡我們通過

flatMap

和

map

的結合，避免了使用for循環，而是直接将每個使用者名下的所有産品挨個傳遞出去，到Subscribe裡處理。

線程的切換

subscribeOn

和

observeOn

RxJava還有個比較牛B的地方，就是可以對線程進行自由的變換。比如主/子線程的切換。

RxJava 提供了

subscribeOn

和

observeOn

來實作線程的切換。可以上事件的産生和處理分布在不同的線程。例如我們可以在子線程擷取網絡資料，在主線程顯示這些資料。

預設情況下如果不指定線程，RxJava的所有操作都是在主線程裡的，在Android開發中如果設計到網絡等操作，是會報異常的，我們可以使用

subscribeOn

和

observeOn

來實作線程的切換，來處理類似的這些情況。

比如我們現在有一個是從伺服器拉取資料的方法叫做

getNetData()

,它的傳回值是String類型，我們都知道在Android中網絡資料的通路是不能在主線程裡進行的，我們隻能在子線程裡進行，然後通過

Handler

發送到主線程。這樣做起來稍微麻煩一些，如果我們使用RxJava就會很簡單。

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            //擷取網絡資料
            String result = getNetData();
            subscriber.onNext(result);
            subscriber.onCompleted();
        }
       //使用subscribeOn(),指定發送事件所在的線程
    }).subscribeOn(Schedulers.newThread())
       //使用observeOn(),指定觀察者接收事件所在的線程
     .observeOn(AndroidSchedulers.mainThread());
           

然後我們根據上面，建立

Subscriber

,然後監聽着

Observable

發送過來的資料就可以了。

本文參考 給 Android 開發者的 RxJava 詳解,感謝扔物線大神