什么是Rx呢？
ReactiveX是Reactive Extensions的缩写，一般简写为Rx.Rx是响应式编程的意思,本质是观察者模式, 是以观察者(Observer)和订阅者(Subscriber)为基础的异步响应方式. 在Android编程时, 经常会使用后台线程, 那么就可以使用这种方式.
RxJava ?
RxJava 主要的作用就是链式完成异步操作,并且非常强大，RxJava最核心的两个东西是Observables（被观察者，事件源）和Subscribers（订阅者）。Observables发出一系列事件，Subscribers处理这些事件。
RxAndroid?
RxAndroid来源于RxJava, 在RxJava的基础上扩展了一些Android的功能.主要的作用也是异步，随意定制主线程、子线程的操作，链式编程，让我们的代码可读性大大增强，总之，是非常好用。
我们可以了解一下官方的开源项目
我们再学习RxAndroid之前肯定是要做一些准备工作的
我们用Android Studio新建项目，直接在build.gradle中添加以下依赖
compile 'io.reactivex:rxandroid:1.2.1'
compile 'io.reactivex:rxjava:1.1.6'
创建观察者
- 定义一个方法，它完成某些任务，然后从异步调用中返回一个值，这个方法是观察者的一部分
- 将这个异步调用本身定义为一个Observable
- 观察者通过订阅（Subscribe）操作关联那个Observable
- 继续你的业务逻辑，等方法返回时，Observable会发射结果观察者的方法会开始处理结果或结果集
观察者回调方法
- 回调方法（onNext、onCompleted、onError）
- onNext( T item)
Observable调用这个方法发射数据，方法的参数就是Observable发射的数据，这个方法可能会被调用多次，取决于你的实现 - onError（Exception e）
当Observable遇到错误或者无法返回期望的数据时会调用这个方法，这个调用会终止Observable，后续不会再调用onNext和onCompleted方法，onError方法的参数书抛出异常 - onCompleted
正常终止，如果没有遇到错误，Observable在最后一次调用onNext之后调用此方法
创建操作Create
- 你可以使用Create操作符从头开始创建一个Observable，给这个操作符传递一个接收观察者作为参数的函数，编写这个函数让它的行为表现为一个Observable，恰当的调用观察者的onNext，onError和onCompleted方法
- 建议你在传递给create方法的函数中检查观察者的isUnsubscribed状态，以便在没有观察者的时候，让你的Observable停止发射数据或者做昂贵的运算
//新建一个RxUtils类
public class RxUtils {
//创建一个观察者方法
public static void createObserable(){
//定义被观察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override public void call(Subscriber<? super String> subscriber) {
if (!subscriber.isUnsubscribed()){
subscriber.onNext("hello");// 发送事件
subscriber.onNext("world");// 发送事件
subscriber.onNext(downLoadJosn());// 发送事件
subscriber.onCompleted();// 完成事件
}
}
});
//订阅者
Subscriber<String> showSub = new Subscriber<String>() {
@Override public void onCompleted() {
Log.i("adu","onCompleted");
}
@Override public void onError(Throwable e) {
Log.i("adu","onError");
}
@Override public void onNext(String s) {
Log.i("adu","onNext"+s);
}
};
observable.subscribe(showSub);//关联被观察者
}
/**
* 创建字符串
*/
public static String downLoadJosn(){
return "json data";
}
}
我们在MainActivity中去创建一个Button的点击事件，然后调用这个方法
会在日志中看到结果是按照顺序依次执行
/**
* create第二种方式
*/
public static void createPrint(){
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override public void call(Subscriber<? super Integer> subscriber) {
if(!subscriber.isUnsubscribed()){
for(int i= 1;i<=10;i++){
subscriber.onNext(i);
}
subscriber.onCompleted();
}
}
}).subscribe(new Subscriber<Integer>() {
@Override public void onCompleted() {
Log.i("adu","onCompleted");
}
@Override public void onError(Throwable e) {
Log.i("adu","onError");
}
@Override public void onNext(Integer integer) {
Log.i("adu","onNext==》》"+integer);
}
});
}
调用方法，查看日志
From操作
将其他种类的对象和数据类型转换为Observable
当你使用Observable时，如果你要处理的数据都可以转换成展现为Observables,而不是需要使用Observables和其他类型的数据
/**
* 使用在被观察者，返回的对象一般都是数值类型
*/
public static void from(){
Integer [] items={1,2,3,4,5,6,7,8,9};
Observable observable = Observable.from(items);
observable.subscribe(new Action1() {
@Override public void call(Object o) {
Log.i("adu",o.toString());
}
});
}
调用方法，查看日志
interval操作
interval操作符既可以延迟执行一段逻辑
/**
* 指定某一时刻进行数据发送
*/
public static void interval(){
Integer [] items = {1,2,3,4,5};
Observable observable = Observable.interval(1,1, TimeUnit.SECONDS);//每隔一秒发送数据
observable.subscribe(new Action1() {
@Override public void call(Object o) {
Log.i("adu",o.toString());
}
});
}
just操作
获取输入数据, 直接分发, 更加简洁, 省略其他回调.
/**
* 处理数组集合
*/
public static void just(){
Integer[] items1={1,2,3,4,5};
Integer[] items2={4,5,1,6,0};
Observable observable = Observable.just(items1,items2);
observable.subscribe(new Subscriber<Integer[]>() {
@Override public void onCompleted() {
Log.i("adu","onCompleted");
}
@Override public void onError(Throwable e) {
Log.i("adu","onError");
}
@Override public void onNext(Integer[] integers) {
for (int i= 0;i<integers.length;i++){
Log.i("adu","onNext==》》"+integers[i]);
}
}
});
}
range操作
range操作符的作用根据出入的初始值n和数目m发射一系列大于等于n的m个值。
/**
* 指定输出数据的范围
*/
public static void range(){
Observable observable = Observable.range(23,4);
observable.subscribe(new Subscriber<Integer>() {
@Override public void onCompleted() {
Log.i("adu","onCompleted");
}
@Override public void onError(Throwable e) {
Log.i("adu","onError");
}
@Override public void onNext(Integer integer) {
Log.i("adu","onNext==》》"+integer);
}
});
}
过滤
/**
* 过滤某些条件
*/
public static void filter(){
Observable observable = Observable.just(1,2,3,4,5,6,7,8);
observable.filter(new Func1<Integer,Boolean>() {
@Override public Boolean call(Integer integer) {
return integer<5;
}
}).observeOn(Schedulers.io()).subscribe(new Subscriber<Integer>() {
@Override public void onCompleted() {
Log.i("adu","onCompleted");
}
@Override public void onError(Throwable e) {
Log.i("adu","onError");
}
@Override public void onNext(Integer integer) {
Log.i("adu","inNext==》》"+integer);
}
});
}
