RxJava的简单应用

时间：2016-06-30 14:40 来源：作者：收藏

问题采用CSS或类似jquery 选择器（selector）语法来处理HTML文档中的数据。方法利用方法：Element.select(String selector)和Elements.select(String selector)。File input = new File(/tmp/input.html);Document doc = Jsoup.parse(

利用业余时间搞了两三个晚上的RxJava,写了两个小demo，特此拿来分享。网上大牛关于RxJava的各种资料到处都有，鉴于水平有限，就不在此过多的对RxJava的各种对象以及操作符做解释了，以免以其昏昏使人昭昭，当然简单的写一下也是可以的。

Observer或者Subscriber：前者是后者的父类，他们也就是RxJava中的观察者，该类中的onNext<T t>方法就是执行业务逻辑的所在，或者用RxJava的术语来说就是处理事件的所在了。

onError:这个方法是事件在传递或者处理的过程中发生错误后会调用到的方法，不管是在Observable还是Observer的onNext方法发生的Exception，RxJava都会调用Observer的onError方法。

就简单的讲这个概念吧，多余的就不在解释，看下面提供的资料了解可以深入的了解。本篇主要介绍RxJava的使用，当然不当之处还请批评指正，共同学习提高，文章的最后会给出demo下载地址！

查阅的资料链接:

《开发技术前线RxJava篇》这个系列文章的主要是RxJava Essentials这本书中的内容，网上可以下载pdf文件进行查看。

《RxAndroid初探》这个也可以瞅瞅，大致的讲解了一下map啦，filter等操作符的应用。

《RxJava官网》学习一个新的东西，当然少不了官网上的权威资料了！

《RxJava Wiki》算是官网提供的api,网上的各种资料也大都是以此为标准进行研究和解说的，可以作为重点参考对象！

《中文API》这个网站也不错，可以对一些重要的api直接查看和使用

开篇之前先说一个有意思的例子，我在单元测试，测试map的时候写了如下代码：

 class MapObserableFactory implements  ObservableFactory{ 
        /**
         * 将事件源转换成另外一个事件源,此处试讲字符串转换成数字
         */
        Func1<String,Integer> func2 = new Func1<String,Integer>(){
            @Override
            public Integer call(String source) {
               return Integer.valueOf(source);
            }
        };

        /**
         * map的作用，把Observable发射的消息转换成另外一个东东
         * @return
         */
        @Override
        public Observable createObservable() {
            return Observable.just("249").map(func2);
        }
    }

上面的代码我是测试map的功能：也就是让Observable发送的数据源转换成另外一个数据源(在次数就是把String 转换成Integer后交给Observer处理)，相应的Observer代码如下：

  class SubscriberFactory implements  ObserverFactory{
        /**
         * Subscriber 是Observer的子类，也是一个观察者
         * @return
         */
        @Override
        public Observer createOberver() {
            Subscriber<Integer> subscriber = new Subscriber<<span style="font-family: Arial, Helvetica, sans-serif;">Integer</span>>() {
                @Override
                public void onCompleted() {
                    Log.d(tag, "OnCompleted");
                }

                //测试发现，onError不仅仅是处理被观察者抛出的错误，当观察者
                //在onNext发生错误的时候，同样会执行该观察者的onError
                @Override
                public void onError(Throwable e) {
                    Log.e(tag,"笨猴子，出错啦");
                }

                @Override
                public void onNext(Integer i) {             
                    i+=1;
                    Log.e(tag,"你这个"+i);//当然输入是：“你这个250”

                }
            };
            return subscriber;
        }
    }

运行一把，完全没问题！不过我刚运行完毕就问自己一个问题：我直接发送“249”的字符串，然后在onNext()里面进行String转成Integer的操作不也是一样的么？何必还要Map操作呢？然后我自己把自己给问住了！不过既然RxJava提供了这种操作总有一定的道理和优点所在，这是必然的！这也可以从另外一种角度来说明一个问题：RxJava允许你在Observable层面对各种事件（数据）进行过滤/转换等操作，然后在发送给指定的Observer对过滤/转换后的事件进行操作；也可以把原始事件直接发送给相应Observer进行处理：比如List集合数据的排序啦等等操作！

说了这么多，开始使用RxJava的第一个例子：

1)获取系统中上已经安装的应用，并且列出应用的名字！

version1.0：

当然是由Observable负责数据的生成和发出了，所以Observable代码如下：

Observable.OnSubscribe onSubscribe = new Observable.OnSubscribe<ResolveInfo>() {
                @Override
                public void call(Subscriber< super ResolveInfo> sub) {
                    List<ResolveInfo> infos=  getAppsInfo();
                    for(ResolveInfo info:infos){                      
                        sub.onNext(info);
                    }
                    sub.onCompleted();
                }

            };
            Observable observable = Observable.create(onSubscribe);

代码很简单，就是在call方法里面执行了三步操作：

1)获取已经安装的应用信息列表

2)遍历此列表，然后调用Observer的onNext事件；或者说就是向Observer发送一条条事件。(注意onNext方法接收的是ResoleInfo对象)

3)数据发送完毕，执行Observer的onCompleted方法

相应的Observer对数据的处理如下：

 Subscriber<ResolveInfo> subscriber = new Subscriber<ResolveInfo>() {
                    @Override
                    public void onCompleted() {

                       showTotalCount();
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(ResolveInfo info) {
                        addView(info);
                    }
                };

上面的Observer做了两种处理：

1)onNext方法把接收的数据通过addView方法添加到一个LinearLayout中展示出来:

 private void addView(ResolveInfo info){
        String appName =   info.loadLabel(getPackageManager()).toString();
        TextView textView = new TextView(this);
        textView.setText(appName);
        linearLayout.addView(textView);
    }

2)在onComplete方法中显示出来系统总共安装多少个应用：

private void showTotalCount(){
        TextView textView = new TextView(this);
        textView.setText("总应用数："+linearLayout.getChildCount() + "条");
        linearLayout.addView(textView);
    }

到此为止，第一个版本关于显示系统应用信息的版本已经产生！

运行界面如下:

RxJava的简单应用

当然上面的Observable需要一个for循环才能发送一个个事件给onNext感觉不是很友好，所以RxJava里面有一个from方法可以供我们使用，所以上面的Observable也可以用如下简单的来代替：

Observable.from(getAppsInfo())

仅仅一行代码就可以代替上面的Observable的for循环来完成一样的功能

version 2.0:

或许你会会说，上面的demo很明显很傻叉，明明有ListView不用干吗非得用LinearLayout这玩意儿呢？所以为了表示高大上，我也用了ListView来处理：重新修改Observer的onNext方法：

   Subscriber<ResolveInfo> subscriber = new Subscriber<ResolveInfo>() {
                    List<ResolveInfo> resolveInfoList = new ArrayList<ResolveInfo>();

                    /**
                     * 数据接收完毕，更新listView
                     */
                    @Override
                    public void onCompleted() {
                        listView.setAdapter(new AppInfoAdapter(resolveInfoList));
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(ResolveInfo info) {
                        resolveInfoList.add(info);
                    }
                };

什么的Observer也做了两步操作：

1)在onNext方法中把接收到的数据保存到一个List集合里面。

2)在onCompleted方法里面向ListView里面添加数据(执行onCompleted方法说明Observable已经将全部数据发送完毕)

version3.0:

完成了第二版本肯定会纳闷，这么一个个的发送数据然后在onNext方法中在一个个的收集数据，有这个必要么，为什么不直接让Observable来发送全部数据集合，Observer的onNext参数接收一个集合处理就是了？为此，第三版正式面世：

Observable的代码修改为如下方式：

            Observable.OnSubscribe onSubscribe = new Observable.OnSubscribe<List<ResolveInfo>>() {
                @Override
                public void call(Subscriber< super List<ResolveInfo>> subscriber) {
                    //Observer方法直接传list数据
                    subscriber.onNext(getAppsInfo());
                    subscriber.onCompleted();
              }
            };

            Observable   observable = Observable.create(onSubscribe);

对应的Observer的onNext方法如下：直接接收Observable接收发送的数据然后交给ListView处理！

  @Override
                    public void onNext(List<ResolveInfo> lists) {

                        listView.setAdapter(new AppInfoAdapter(lists));
                    }

当然这个版本的Observable也代码也可以改成一行如下的代码

Observable.from(getAppsInfo()).toList();

到此为止通过RxJava来实现获取系统已经安装app的简单demo已经编写完毕！下面开始说RxJava的第二个应用：

文件下载，并且监听下载进度然后更新下载进度条

前段时间刚在项目中做了下载安装并且更新下载进度条的逻辑，所以在学习RxJava的时候准备用它来实现一些这个功能。下载文件是个耗时的操作，肯定需要在非UI线程中操作；而跟新进度条是在UI线程中使用，所以简单的Observable和Observer配置如下：

  subscription= observableFactory.createObservable()
                .subscribeOn(Schedulers.io())//生产事件在io
                .observeOn(AndroidSchedulers.mainThread())//消费事件在UI线程
                .subscribe(subscriberFactory.createOberver());

在这里我们让Observable在io线程进行下载，下面看看Observable的具体在做了神马：

 Observable.OnSubscribe onSubscribe = new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber< super Integer> subscriber) {
                downloadFile(subscriber);
            }
        };
        observable =  Observable.create(onSubscribe);

call方法中调用了downloadFile来执行下载文件的操作，并把subscriber这个Observer传入到downloadFile方法里面:

 public void downloadFile(Subscriber< super Integer> subscriber) {
        InputStream input = null;
        OutputStream output = null;
        HttpURLConnection connection = null;
        try {
            URL url = new URL(source);
            connection = (HttpURLConnection) url.openConnection();
            connection.connect();
            if (connection.getResponseCode() != HttpURLConnection.HTTP_OK) {
                subscriber.onError(new Exception("网络不正常"));
                return;
            }
            int fileLength = connection.getContentLength();
            input = connection.getInputStream();
            output = new FileOutputStream(savePath);
            byte data[] = new byte[4096];
            long total = 0;
            int count;
            subscriber.onStart();       
            while ((count = input.read(data)) != -1) {
                if(subscriber.isUnsubscribed()){//如果已经取消订阅，减少不必要的数据生成项
                    break;
                }
                total += count;
                //当前下载的进度
                int currentProgress =(int) (total*100/fileLength);
                subscriber.onNext(currentProgress);
                 
                output.write(data, 0, count);
            }//end while
            subscriber.onCompleted();
        } catch (Exception e) {
            subscriber.onError(e);
        } finally {
            try {
                if (output != null) {
                    output.close();
                }
                if (input != null) {
                    input.close();
                }
            } catch (IOException e) {
                subscriber.onError(e);
            }
            if (connection != null) {
                connection.disconnect();
                subscriber.onCompleted();
            }
        }
    }

那么在Observer的noNext方法执行的更新进度条的逻辑就是简单的如下操作了：

 Subscriber<Integer> subscriber = new Subscriber<Integer>() {
                @Override
                public void onStart() {

                    progressBar.setVisibility(View.VISIBLE);
                }

                @Override
                public void onCompleted() {

                    Log.i(tag,"下载结束");
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(tag,"下载出错");
                   e.printStackTrace();
                }

                @Override
                public void onNext(Integer progress) {
                    Log.i(tag,"下载...."+progress);
                    progressBar.setProgress(progress);
                }
            };

运行一把，很遗憾在切换大文件下载的时候偶尔会运行出错，错误如下：

RxJava的简单应用

在Wilki里面查到了关于这个Backpressure错误的说明：

In RxJava it is not difficult to get into a situation in which an Observable is emitting items more rapidly than an operator or subscriber can consume them

就是Observable发送事件的速度大于Observer处理事件的速度；简单的来说就是Observer这个观察者累死了！并且该片还说明了解决这个问题的几种方式，不过正如文章所说：Backpressure doesn’t make the problem of an overproducing Observable or an underconsuming Subscriber go away. It just moves the problem up the chain of operators to a point where it can be handled better.这种错误是不可能根除的，只能提供一些方法来来让Observable发送的事件通过一些方法减缓一下，比如通过onBackpressureBuffer来在一个合适的时机来发送一些数据；不过这种监听下载进度的当然是实时来进行更新，所以关于Backpressure的提供的几个方法再本篇中不能通过；于是我在跟android群里的小伙伴沟通过后，进行排查。发现在下载文件的while循环如下：

 while ((count = input.read(data)) != -1) {
                if(subscriber.isUnsubscribed()){//如果已经取消订阅，减少不必要的数据生成项
                    break;
                }
                total += count;
                //当前下载的进度百分比
                int currentProgress =(int) (total*100/fileLength);
                System.out.println("currentProgress=="+currentProgress);          
                 subscriber.onNext(currentProgress);

                output.write(data, 0, count);
            }//end while

我在while循环里面添加了打印，打印信息如下：

RxJava的简单应用

这还只是一部分，打印了一大堆的0%或者1%等，好几屏幕的打印都是它们！！！！这些说明进度百分比没有变化的时候在上面的while循环中onNext方法也会执行，这些是没有必要的。针对此情况对while我做了如下的优化：记录上次下载的进度，用下一次加载的进度进行相减，如果大于零说明更新百分比有变动，那么就执行onNext方法，这样就可以避免如上面打印的那种情况下执行onNext情况的出现!

修改后的代码如下：

 int preProgress = 0;//记录之前的进度
            while ((count = input.read(data)) != -1) {
                if(subscriber.isUnsubscribed()){//如果已经取消订阅，减少不必要的数据生成项
                    break;
                }
                total += count;
                //当前下载的进度
                int currentProgress =(int) (total*100/fileLength);
                System.out.println("currentProgress=="+currentProgress);
                if(currentProgress-preProgress>0){
                    subscriber.onNext(currentProgress);
                    preProgress = currentProgress;
                }

                output.write(data, 0, count);
            }//end while

然后在运行，下载几次，上面的那个错误就没出现了!

到此为止，对于RxJava的简单应用就写完了，本篇并没有对RxJava的相关内容进行详细深入的研究，只是简单的用了一下RxJava，想要深入了解的话可参阅文章开头部分的列举的几个网站进行学习。文中demo下载地址(点此下载），demo代码组织的不是很好，不当的地方欢迎批评指正，共同学习和提高！