RxJava入门

要理解RxJava，首先得理解什么是（异步）数据流。一些典型的点击事件本质上就是一个异步数据流，这样你就可以观察它的变化并使其做出一些反应。数据流无处不在，任何东西都可以转变为数据流，例如变量，用户输入，属性，缓存，数据结构等等，每天使用的微博，你可以把微博订阅想象成点击事件一样的数据流，我们可以监听这样的数据流，并作出相应的反应。

一个数据流可以作为另一个数据流的输入，甚至多个数据流也可以作为另一个数据流的输入。你可以合并两个数据流，也可以过滤一个数据流得到另一个只包含你感兴趣的事件的数据流，还可以映射一个数据流的值到一个新的数据流里。

一个数据流是一个按时间排序的即将发生的事件(Ongoing events ordered in time)的序列。如上图，它可以发出3种不同的事件：一个某种类型的值事件，一个错误事件和一个完成事件。当一个完成事件发生时，在某些情况下，我们可能会做这样的操作：关闭包含那个按钮的窗口或者视图组件。

我们只能异步捕捉被发出的事件，使得我们可以在发出一个值事件时执行一个函数，发出错误事件时执行一个函数，发出完成事件时执行另一个函数。有时候你可以忽略后两个事件，只需聚焦于如何定义和设计在发出值事件时要执行的函数，监听这个事件流的过程叫做订阅，我们定义的函数叫做观察者，而事件流就可以叫做被观察的主题(或者叫被观察者)。它就是观察者模式。

了解了以上知识，你还得了解什么是观察者模式，至于审是观察者模式，请自行查阅相关信息，你可以简单的理解为很多人订阅他们感兴趣的事物，而他们感兴趣的事物会在合适的时间通知他们。
我们编写一个RxJava的例子。

 Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                try{
                    if (!subscriber.isUnsubscribed()){
                        subscriber.onStart();
                        for (int i=0;i<5;i++){
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                }catch (Exception e){
                    subscriber.onError(e);
                }
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onStart() {
                System.out.println("start");
            }

            @Override
            public void onCompleted() {
                System.out.println("complete");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println(e.getMessage());
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println(integer);
            }
        });

我们通过create函数创建了一个被观察者，被观察者会发出一些数据流，如我们调用了onStart，onNext，onCompleted，onError等函数，这些函数会在观察者中回调。在我们的被观察者中，我们首先通知观察者我们开始发送他们感兴趣的东西，这里是Integer整形数据，之后连续发送了5个整形数据，发送完成后调用了onCompleted通知数据发送完成，而这个过程一旦发送什么错误，onCompleted将不会被调用，代替它的就是通知观察者数据发送过程中发生了错误，也就是onError的调用。

或许你对错误事件和完成事件不感兴趣，那么你可以使用subscribe的重载函数。就像这样子

Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                try{
                    if (!subscriber.isUnsubscribed()){
                        subscriber.onStart();
                        for (int i=0;i<5;i++){
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                }catch (Exception e){
                    subscriber.onError(e);
                }
            }
        }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });

同样的，被观察者也可以简单的通过just函数进行创建

 Observable.just(0,1,2,3,4).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });

甚至你可以使用Lambda表达式简化代码

Observable.just(0,1,2,3,4).subscribe(System.out::println);

就这样，我们完成了一个Helloworld。RxJava的学习曲线有点陡峭，但是只要你理解了什么是观察者模式，什么是异步数据流，以及你理解了其操作符，一切还是简单的。这篇文章也只是个入门，RxJava的东西还有很多灯我们去学习。

参考文章
那些年我们错过的响应式编程