前言

　　RxAndroid是RxJava在Android上的一个扩展，大牛JakeWharton的项目。据说和Retorfit、OkHttp组合起来使用，效果不是一般的好。而且用它似乎可以完全替代eventBus和OTTO，这么牛的东西当然要研究研究了 ，看看它到底有多厉害。
正文
相关资源

　　RxJava的GitHub地址：https://github.com/ReactiveX/RxJava 
　　RxAndroid的GitHub地址：https://github.com/ReactiveX/RxAndroid 
　　中文文档：https://mcxiaoke.gitbooks.io/rxdocs/content/ 
　　一篇写的比较好的入门RxJava的文章的地址：http://gank.io/post/560e15be2dca930e00da1083
1.RxJava是干嘛的

　　Rx(Reactive Extensions)是一个库，用来处理事件和异步任务，在很多语言上都有实现，RxJava是Rx在Java上的实现。简单来说，RxJava就是处理异步的一个库，最基本是基于观察者模式来实现的。通过Obserable和Observer的机制，实现所谓响应式的编程体验。 
　　Android的童鞋都知道，处理异步事件，现有的AsyncTask、Handler，不错的第三方事件总线EventBus、OTTO等等都可以处理。并且大部分童鞋应该都很熟练了。而且经我目前的学习来看，RxJava这个库，上手确实有门槛，不是拿来就能用。但是作为一个猿，那些可能出现的优秀的框架技术，及时的跟进和学习是必要的，从中汲取营养才能帮助自己成长。况且有童鞋已经表示，它完全可以替代EventBus和OTTO，来看看吧。
2.RxJava的优势

　　最概括的两个字：简洁。而且当业务越繁琐越复杂时这一点就越显出优势——它能够保持简洁。 
　　简单的demo看不出来，真正投入项目使用了应该就有体会了。它提供的各种功能强悍的操作符真的很强大。
3.基本使用流程

　　这里只介绍Android Studio的接入方式，如果你还在用Eclipse的话，我建议你换了。 
　　配置buile.gradle:(以下为当前最新版本，如有更新请到上述GitHub链接查看更新)

    dependencies {
    　　compile ‘io.reactivex:rxandroid:1.2.1‘
    　　compile ‘io.reactivex:rxjava:1.1.6‘
    　　}


　　配置完之后就可以使用RxJava的API了。介绍两个个关键的类： 
　　(1)Observable (2)Subscriber 即：被观察者(Observable)和观察者(Subscriber)，其实我觉得叫发布者和订阅者更好理解一些，但大家都叫被观察者和观察者。 
　　主干的使用过程就是1.创建被观察者。2.创建观察者。3.将二者建立联系。完毕。然后被观察中发出信息触发观察者的动作，执行相应的方法，就这样。你先别急着吐槽它很平庸。它的强大在于这个过程中提供的各种操作变换的技巧会让你可以简洁的处理相当繁琐的代码逻辑。 
先看一个简单的demo:

            //创建一个被观察者(发布者)
            Observable observable = Observable.create(new Observable.OnSubscribe<Integer>() {
                @Override
                public void call(Subscriber<? super Integer> subscriber) {
                    subscriber.onNext(1001);
                    subscriber.onNext(1002);
                    subscriber.onNext(1003);
                    subscriber.onCompleted();
                }
            });
     
            //创建一个观察者
            Subscriber<Integer> subscriber = new Subscriber<Integer>() {
                @Override
                public void onCompleted() {
                    Log.d(TAG, "onCompleted.. ");
                }
     
                @Override
                public void onError(Throwable e) {
                    Log.d(TAG, "subscriber onError.. "　+　e.getMessage());
                }
     
                @Override
                public void onNext(Integer integer) {
                    Log.d(TAG, "onNext.. integer:" + integer);
                }
            };
     
            //注册观察者(这个方法名看起来有点怪，还不如写成regisiterSubscriber(..)或者干脆addSubscriber(..))
            //注册后就会开始调用call()中的观察者执行的方法 onNext() onCompleted()等
            observable.subscribe(subscriber);

  

　　上面的例子中，当Observable发射数据时，会依次调用Subscriber的onNext()方法，将发射的数据作为参数传给onNext()，如果出错，则会调用Subscriber的onError()方法，完成所有数据发射后，调用onCompleted()方法，整个过程完毕。 
　　但是，subcribe()方法默认在当前线程被调用。所以，这样使用的话，被观察者和观察者的所有的动作都是在同一个线程完成的，没卵用… 
　　但是当然肯定不会就这个程度了，RxJava有两个方法可以很方便的指定观察者和被观察者代码运行的线程，RxAndroid还有一个扩展，可以指定在UI线程运行。你懂的！ 
如下：

    //设置观察者和发布者代码所要运行的线程后注册观察者
    observable.subscribeOn(Schedulers.immediate())//在当前线程执行subscribe()方法
    .observeOn(AndroidSchedulers.mainThread())//在UI线程执行观察者的方法
    .subscribe(subscriber);

  

　　通过Scheduler作为参数来指定代码运行的线程，非常方便，好用到不行…其他常用的参数还有： 
　　Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。　 
　　Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。 
　　Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免建不必要的线程。 
　　Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。 
　　另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
4.Observable创建方式

　　以上介绍了主干使用流程，从这里我们往细一点再看。前文说了，RxJava的强大之处在于它的各种操作符。在创建Observable对象的方式上，同样有很多方便的操作符的实现，上面是通过Observable.create()方法创建的observable对象，这里介绍其他几个常用的方法。 
　　 
　　通过from创建Observable：

    　　　　　//Teacher为一个数据Bean，包含姓名，年龄，住址三个字段
    　　　　　List<Teacher> teachers = new ArrayList<>();
            for (int i = 0; i < 4; i++) {
                teachers.add(new Teacher("name" + i, i, "place" + i));
            }
            //from方法支持继承了Interable接口的参数，所以常用的数据结构(Map、List..)都可以转换
            Observable fromObservale = Observable.from(teachers);
            fromObservale.subscribe(new Subscriber<Teacher>() {
                @Override
                public void onCompleted() {
                    Log.i(TAG, "from(teachers)  onCompleted");
                }
     
                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "from(teachers)  " + e.getMessage());
                }
     
                @Override
                public void onNext(Teacher teacher) {
                    //依次接收到teachers中的对象
                    Log.d(TAG, "from(teachers)  onNext:" + teacher.toString());
                }
            });

  

　　用from方法创建Observable，可以传入一个数组，或者一个继承了Iterable的类的对象作为参数，也就是说，java中常用的数据结构如List、Map等都可以直接作为参数传入from()方法用以构建Observable。这样，当Observable发射数据时，它将会依次把序列中的元素依次发射出来。 
　　 
　　通过just创建Observable：　　

    //Just类似于From，但是From会将数组或Iterable的元素具取出然后逐个发射，而Just只是简单的原样发射，将数组或Iterable当做单个数据。
    //Just接受一至九个参数，返回一个按参数列表顺序发射这些数据的Observable
    Observable justObservable = Observable.just(1, "someThing", false, 3.256f, new Teacher("Jhon", 25, "NewYork"));
    justObservable.subscribe(new Subscriber() {
        @Override
        public void onCompleted() {
            Log.i(TAG, "just(...)  onCompleted");
        }
     
        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "just(...)  onError:" + e.getMessage());
        }
     
        @Override
        public void onNext(Object o) {
            Log.d(TAG, "just(...)  onNext:" + o.toString());
        }
    });

  
　　just直接接收object作为参数，原样发射出来，也是非常方便的。 
　　 
通过timer创建Observable：　　

    //timer()创建一个Observable，它在一个给定的延迟后发射一个特殊的值 设定执行方法在UI线程执行
    //延时两秒后发射值
    //实测 延时2s后发送了一个0
    Observable timerObservable = Observable.timer(2, TimeUnit.SECONDS, AndroidSchedulers.mainThread());
       timerObservable.subscribe(
               new Subscriber() {
                   @Override
                   public void onCompleted() {
                       Log.i(TAG, "timer(...)  onCompleted");
                       refreshStr("timer(...)  onCompleted\n");
                   }
     
                   @Override
                   public void onError(Throwable e) {
                       Log.e(TAG, "timer(...)  onError:" + e.getMessage());
                       refreshStr("timer(...)  onError:" + e.getMessage());
                   }
     
                   @Override
                   public void onNext(Object o) {
                       Log.d(TAG, "timer(...)  onNext:" + o.toString());
                       refreshStr("timerObservable 延时两秒触发 发送值：" + o.toString());
                   }
               }
       );

  

　　timer有定时的作用，延时发送一个值０。 
　　 
通过range创建Observable(这里叠加使用一个repeat方法)：　　

    //range 发射从n到m的整数序列 可以指定Scheduler设置执行方法运行的线程
    //repeat方法可以指定重复触发的次数
    Observable rangeObservable = Observable.range(3, 7).repeat(2);
    rangeObservable.subscribe(
    //在不写观察者的情况下，可以使用Action1和Action0这两个接口来实现不完整定义的回调； 参见：ActionSubscriber
    //Action1<T>可以代替实现onNext(); Action1<Throwable>可以代替实现onError(); Action0可以代替实现onConplete()
            new Action1() {
                @Override
                public void call(Object o) {
                    Log.e(TAG, "range(3, 7).repeat(2)  onNext:"+o.toString());
                }
            },
            new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    Log.e(TAG, "range(3, 7).repeat(2)  "+throwable.getMessage());
                }
            },
            new Action0() {
                @Override
                public void call() {
                    Log.i(TAG, "range(3, 7).repeat(2)  onCompleted");
                }
            });

  

　　range发射从n到m的整数序列，repeat可以指定重复次数，以上发射的次序为：3，4，5，6，7，3，4，5，6，7。这里用到的Action0和Action1是两个可以替代Subscriber的接口，具体可以参见相关文档和源码实现，这里不深入介绍。 
　　其他还有Interval、Defer、Start等方法就不一一介绍了，本文主要是帮助初次接触的童鞋入门，RxJava的操作符非常丰富，这里很难一一说明，更多的内容要还需要大家自己去熟悉和探究。
５.变换操作

　　除了多样的Observable创建方式，RxJava还有一个神奇的操作就是变换。通过自己定义的方法，你可以将输入的值变换成另一种类型再输出(比如输入url，输出bitmap)，单一变换、批量变换、甚至实现双重变换，嵌套两重异步操作！并且代码格式一如既往的干净平整。是不是很牛？ 
　　 
　　使用map()方法做转换:

    Runnable run = new Runnable() {
         @Override
         public void run() {
             //将文件路径转换为bitmap发出 观察者直接收到bitmap进行处理
             Observable observable = Observable.just(imgFilePath);
             observable.map(new Func1<String, Bitmap>() {
                 @Override
                 public Bitmap call(String imgFilePath) {
                     return getBitmapFromAssets(imgFilePath);
                 }
             }).subscribeOn(Schedulers.immediate())//当前线程(子线程)发布
                     .observeOn(AndroidSchedulers.mainThread())//UI线程执行(更新图片)
                     .subscribe(new Subscriber<Bitmap>() {
                         @Override
                         public void onCompleted() {
                             Log.i(TAG, "observable.map(..)  onCompleted");
                         }
     
                         @Override
                         public void onError(Throwable e) {
                             Log.i(TAG, "observable.map(..)  onError" + e.getMessage());
                         }
     
                         @Override
                         public void onNext(Bitmap bitmap) {
                             //显示图片
                             iv.setImageBitmap(bitmap);
                         }
                     });
         }
     };
     new Thread(run).start();



　　map()方法是最基本的变换操作，这里只变换了一个数据，将文件路径解析为Bitmap显示出来。你当然也可以多传入几个参数或者用from操作符传入一个数组或者集合等，批量操作，并且同时指定代码运行的线程。而且这些所有的操作都可以在一条链式代码中全部完成，易读易维护。你是不是已经有一点体会到它的威力了？ 
　　 
　　flatMap()实现双重变换 
　　flatMap()将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable。即：第一次转换时，它依次将输入的数据转换成一个Observable，然后将这些Observable发射的数据集中到一个Observable里依次发射出来。觉得莫名其妙？来看一个实际例子：

    Subscriber subscriber = new Subscriber<Integer>() {
        @Override
        public void onCompleted() {
            Log.i(TAG,"Observable.just(array1,array2).flatMap  onCompleted\n\n");
        }
     
        @Override
        public void onError(Throwable e) {
            Log.e(TAG,"Observable.just(array1,array2).flatMap   onError  "+e.getMessage());
        }
     
        @Override
        public void onNext(Integer integer) {
            Log.d(TAG,"Observable.just(array1,array2).flatMap  integer = "+integer);
        }
    };
     
    //flatMap可以实现一个双重转换，在它的回调方法中会返回一个observable对象，但它并不会直接发射这个对象
    //而是将这个observable对象要发射的值 集中到一个新的observable对象中依次发射
    //如本例，第一层Observable依次发射两个数组，经过flatmap转换之后，变成变成两个依次发射数组元素的observable
    // 最后在subscriber中接收到的直接是整型数，等于将两个数组"铺开"了，直接发射整数，这就是大概地"flat"的含义吧
    // flatMap方法可以很灵活的使用，实现双重变换，满足很多不同情况下的需求,比如处理嵌套的异步代码等，非常棒!
    Integer[] array1 = {1, 2, 3, 4}, array2 = {5, 6, 7, 8};
    Observable.just(array1,array2).flatMap(new Func1<Integer[], Observable<?>>() {
        @Override
        public Observable<?> call(Integer[] ints) {
            Observable observable = Observable.from(ints);
            return observable;
        }
    }).subscribe(subscriber);

    

　　这里flatMap()方法将最初传入的两个数组在第一次变换时，通过from操作符变换成两个Observable，然后在将这两个Observable发射的数据全部集中到一个新的Observable中集中发射，等于将两个数组”铺开”了，依次发射出来它们的元素。具体转换的方法由你指定，使用的方式是比较灵活的。 
　　比如有的商城类应用的需求：先要拿到某类别的一个产品列表，然后列表中有具体产品展示图片的url，需要你拿到产品列表信息后依次去请求图片，成功后更新到UI页面上，使用flatMap，你肯定知道怎么写了吧，是不是比CallBack跳来跳去的舒服一些？

　　scan()变换 
　　scan操作符对原始Observable发射的第一项数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。它持续进行这个过程来产生剩余的数据序列。 
　　当看到这里的时候，我已经由衷的在感叹，这些操作符实在太TM丰富了，然后对它的强大已经开始有所体会和感悟了。这种产生斐波那契数列的操作都给封装进去了，而且函数由你自定义，你能用它做成什么，可能在灵感到来之前你自己都想不到。 
　　demo代码：

    //scan 会将输入的第一个元素当作参数做一个函数运算(函数由你实现，规定需要两个参数，此时另一个默认没有)，然后发射结果
    // 同时，运算结果会被当作函数的与第二个参数与第二个元素再进行函数运算，完成后发射结果
    // 然后将这个结果与第三个元素作为函数的参数再次运算...直到最后一个元素
    Observable.just(1, 2, 3, 4).scan(new Func2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer integer, Integer integer2) {
            //integer是第一个元素或上一次计算的结果，integer2是下一轮运算中新的序列中元素
            Log.d(TAG, "scan call   integer:" + integer + "   integer2:" + integer2);
            return integer + integer2;
        }
    }).subscribe(new Subscriber<Integer>() {
        @Override
        public void onCompleted() {
            Log.i(TAG, "Observable.just(1,2,3,4).scan   onCompleted..");
            initViewState();
        }
     
        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "Observable.just(1,2,3,4).scan  onError  " + e.getMessage());
        }
     
        @Override
        public void onNext(Integer integer) {
            Log.d(TAG, "Observable.just(1,2,3,4).scan  onNext()..  integer = " + integer);
            /**
             * 第一次为1，然后是3(1+2)，6(3+3),10(6+4)
             */
        }
    });

    

　　注释和上面的说明都很清晰了，就不再赘述。同样，关于转换操作，也还有很多其他的操作符，如wiindow() buffer() 等已实现的方法，具体参见文档吧。
6.过滤、结合操作

　　在文档的分类中，还有两片基础API是过滤和结合的操作符，例如:Filter、Skip、Take、Ｍerage、Zip等等，本来打算一起列举的，但是想想其实如果熟悉了上面的内容，这两块相关的API上手其实也很容易了。如果入门目的已经达到，再讲这个显得有点啰嗦。所以略去，如果以后有心得，在开篇另讲那些操作符的使用，RxJava的知识，主要还是要靠自己多熟悉，多研究。
尾声

　　本文主要希望帮助初次接触RxJava的童鞋入门，讲的一些基础知识，RxJava太大，内容太丰富，入门之后需要下的功夫也不少，希望大家都能day day up! 
　　最后，如发现内容有误，请斧正！ 
　　非常感谢！