什么是RxJava
RxJava 在 GitHub （https://github.com/ReactiveX/RxJava）上的介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。
其实， RxJava 的本质可以简单用异步这个词概括，它就是一个实现异步操作的库。
基本概念
RxJava里，我们有四种主要角色：
Observable：发射源，被观察者
Observer：接收源，观察者
Subscriber：订阅者，也是接收源，跟Observer之间的区别在于Subscriber实现了Observer接口，比Observer多了一个最重要的方法unsubscribe( )，用来取消订阅，当你不再想接收数据了，可以调用unsubscribe( )方法停止接收，还多了一个方法是onStart()，它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。另外Observer 在 subscribe() 过程中,最终也会被转换成 Subscriber 对象。
Subject：Subject既可充当发射源，也可充当接收源。
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。
还有一些基本概念如下：
Subscription ：Observable调用subscribe( )方法返回的对象，同样有unsubscribe( )方法，可以用来取消订阅事件；
Action0：RxJava中的一个接口，它只有一个无参call（）方法，且无返回值，同样还有Action1，Action2...Action9等，Action1封装了含有 1 个参的call（）方法，即call（T t），Action2封装了含有 2 个参数的call方法，即call（T1 t1，T2 t2），以此类推；
Func0：与Action0非常相似，也有call（）方法，但是它是有返回值的，同样也有Func0、Func1...Func9;
回调方法
Subscribe方法用于将观察者连接到Observable，你的观察者需要实现以下方法：
onNext(T item)
Observable调用这个方法发射数据，方法的参数就是Observable发射的数据，这个方法可能会被调用多次，取决于你的实现。
onError(Exception ex)
当Observable遇到错误或者无法返回期望的数据时会调用这个方法，这个调用会终止Observable，后续不会再调用onNext和onCompleted，onError方法的参数是抛出的异常。
onComplete
正常终止，如果没有遇到错误，Observable在最后一次调用onNext之后调用此方法。
案例
先创建个数据发射源：
Observable sender = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext("Hello！"); //发送数据"Hello！"
}
});
再创建个数据接收源：
Observer receiver = new Observer() {
@Override
public void onCompleted() {
//数据接收完成时调用
}
@Override
public void onError(Throwable e) {
//发生错误调用
}
@Override
public void onNext(String s) {
//正常接收数据调用
System.out.print(s); //将接收到来自sender的问候
}
};
将发射源和接收源关联起来：
sender.subscribe(receiver);
别忘了添加依赖：
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
基本用法
创建 Observer
除了 上述例子的Observer 接口之外，RxJava 还内置了一个实现了 Observer 的抽象类：Subscriber。 Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的：
Subscriber subscriber = new Subscriber() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
创建Observable
（1）使用create( )
normalObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext("create1"); //发射一个"create1"的String
subscriber.onNext("create2"); //发射一个"create2"的String
subscriber.onCompleted();//发射完成,这种方法需要手动调用onCompleted，才会回调Observer的onCompleted方法
}});
（2）使用just( )，它接受一至九个参数，返回一个按参数列表顺序发射这些数据的Observable。
justObservable = Observable.just("just1","just2"); //依次发送"just1"和"just2"
（3）使用from( )，接受数组，返回一个按参数列表顺序发射这些数据的Observable。
Listlist = new ArrayList<>();
list.add("from1");
list.add("from2");
list.add("from3");
fromObservable = Observable.from(list); //遍历list 每次发送一个
（4）使用defer( )，有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable。
deferObservable = Observable.defer(new Func0>() {
@Override
//注意此处的call方法没有Subscriber参数
public Observable<String> call() {
return Observable.just("deferObservable");
}
});
(5)使用range( )，创建一个发射特定整数序列的Observable，第一个参数为起始值，第二个为发送的个数，如果为0则不发送，负数则抛异常。
rangeObservable = Observable.range(10, 5); //将发送整数10，11，12，13，14
（6）使用timer( )，创建一个Observable，它在一个给定的延迟后发射一个特殊的值，等同于Android中Handler的postDelay( )方法。
timerObservable = Observable.timer(3, TimeUnit.SECONDS); //3秒后发射一个值
（7）使用interval( )，创建一个按固定时间间隔发射整数序列的Observable，可用作定时器。
intervalObservable = Observable.interval(1, TimeUnit.SECONDS); //每隔一秒发送一次
（8）使用repeat( )，创建一个重复发射特定数据的Observable。
repeatObservable = Observable.just("repeatObservable").repeat(3); //重复发射3次
变换操作
Map
操作符对原始Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果。如下，将原始Observable数据转化成小写，再发射：
Observable.just("Hello", "RxJava")
.map(new Func1() {
@Override
public String call(String s) {
return s.toLowerCase();
}
})
.subscribe(new Observer() {
@Override
public void onCompleted() {
Log.d(tag, "onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d(tag, "onNext=" + s);
}
});
网友评论