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 详解,感谢扔物线大神