- 如果问移动开发的潮流是什么的话，那么在这个16年初的移动技术圈里 RxJava（响应式编程）+Retrofit（可能是目前最好的注解式网络框架）+ORM（当然如果你信任完全面向对象的数据库可以看一下Realm（不是ORM）据说是目前最好的面向对象数据库全系统解决方案了）
本文记录一些本屌学习RxJava/RxAndroid响应式编程的内容
-
什么是RxJava？为什么要使用Rxjava？
官方：RxJava 是一个在Java虚拟机上实现的响应式扩展库：提供了基于observable序列实现的异步调用及基于事件编程。 它扩展了观察者模式，支持数据、事件序列并允许你合并序列，无需关心底层的线程处理、同步、线程安全、并发数据结构和非阻塞I/O处理。
简单来说：就是一个使得异步处理起来草鸡简单的解决方案。
项目中有没有Thread的漫天飞；一用AsyncTask又感觉很多简单功能立马复杂了；Handler数据传递不好跟踪，处理的事物又非常的重？
那么RxJava简直就是救星了！！
网上很多在介绍RxJava的文章为了体现笔者高端，都是用Lambda表达式来演示的，因为Lambda的可读性其实是很不好的，我很不推荐初学者使用Lambda来使用RxJava，这样使用会很不理解RxJava本身的，所以本文就不附带Lambda了
BUT
没有什么框架是全能的，那么Rxjava的缺点是什么呢？
总结：
Using RxJava for individual tasks is low risk and potentially high gain. It can simplify your code considerably. However, the more you use Rx in your project, the more likely you are to see a domino effect of a reactive expansion. Bridging the gap between non-Rx and Rx parts can be troublesome, and it can become tempting to simply write everything in Rx. This, on the other hand, is a decision not easily made. While indescribably deliberating, you will be wandering into a new land with unthinkable possibilities but with very little help.
我的理解就是：
- 在轻量级的应用开发中RxJava能够提供隔壁高性能和效率的，但是在很大的项目中使用RxJava的话，可能会有一些引用或者内存泄露的问题产生。
- 并且RxJava自发布到现在已经超过两年了，API也一直有不小的变动，需要开发者一直关注Rxjava的变更。
当然这并不能妨碍我们对这一框架的膜拜和学习，下面进入正题...
-
RxJava的概念们
RxJava主要有四个主体概念：Scheduler(线程控制)、Observable (被观察者)、Observer (观察者)以及Subscribe (订阅、事件)
1.Scheduler(线程控制)
Scheduler负责线程调度，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ，它们已经适合大多数的使用场景：
Schedulers.computation() 用于计算任务，如事件循环或和回调处理，不要用于IO操作(IO操作请使用Schedulers.io())；默认线程数等于处理器的数量
Schedulers.from(executor) 使用指定的Executor作为调度器
Schedulers.immediate( ) 在当前线程立即开始执行任务
Schedulers.io() 用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长；对于普通的计算任务，请使用Schedulers.computation()
Schedulers.newThread() 为每个任务创建一个新线程
Schedulers.trampoline() 当其它排队的任务完成后，在当前线程排队开始执行
另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
Observable通过使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制。
- subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
- observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
Observable.create(new Observable.OnSubscribe(){...})//创建一个订阅获取网络数据的Observable
.subscribeOn(Schedulers.io())// 在非UI线程中执行获取网络数据
.observeOn(AndroidSchedulers.mainThread())// 在UI线程中执行结果
写个获取网络数据的伪代码
Observable.create(new Observable.OnSubscribe<NewsBean>() {
@Override
public void call(Subscriber<? super NewsBean> subscriber) {
//获取网络数据
NewsBean bean = loadDataFromNet();
subscriber.onNext(bean);
}
})
.subscribeOn(Schedulers.io())// 在非UI线程中执行获取网络数据
.observeOn(AndroidSchedulers.mainThread())// 在UI线程中执行结果
.subscribe(new Observer<NewsBean>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(NewsBean newsBean) {
//更新UI上面的数据
mViewHolder.bindData(newsBean);
}
});
是不是没有Thread，没有handler，没有AsyncTask
2.Observable (被观察者)
Observable可以说是使用的核心了，熟悉了Observable的API基本上就了解了RxJava所能帮你做那些事情了，先来看一下Observable中API们，其实Observable这个类本身就是Observable的一个生成器模式，提供了大量的在不同场景下面创造一个被观察者的方法。
- 下面来看一个最简单的创建Observable的方法
Observable observable;
observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("onStart");
subscriber.onNext("onDestroy");
//onCompleted 和 onError都是终止调用，只会被出发之一
subscriber.onCompleted();
subscriber.onError(new Throwable("onError"));
}
});
是不是很好理解，就是创建一个带有OnSubscribe的被观察者，OnSubscribe接口里面的call方法是在当observable.subscribe的时候被调用（同OnSubscribe的字面意思：在被订阅的时候做的事）
- 而且Observable提供了非常多的创建方法（也就是其他很多文章中的“操作符”）,下面列举一下Observable常用的方法和介绍
subscribe
创建一个观察者观察当前事件，并且调用当前被观察者的call()方法
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("onStart");
subscriber.onNext("onDestroy");
}
}).subscribe(new Observer<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
println(s);
}
});
//结果就是打印出了
//onStart
//onDestroy
just & from
静态方法用来创建了一个Observable：顺序调用subscriber.onNext执行定义的数组
Observable<String> observable = Observable.just("onStart","onDestroy");//完全等同于上面的create方法
//同等
String[] operations = {"onStart","onDestroy"};
Observable observable = Observable.from(operations);
**map **
创建一个新的Observable包装一个现有的Observable，使得每个到达观察者的数据是新的Observable来的。
Observable<String> observable = Observable.just("onStart","onDestroy");
Observable mapObservable = observable.map(new Func1<String, String>() {
@Override
public String call(String s) {
return "activity " + s;
}
});
//那么到达mapObservable的观察者的数据就是("activity onStart","activity onDestroy")
take
限制观察者的onNext方法调用的次数
Observable<String> observable = Observable.just("onStart","onDestroy");
Observable<String> takeObservable = observable.take(1);
//这样到达takeObservable的观察者的数据就只有（"onStart"）了
Observable的方法先简单介绍到这把，后面讲Observable原理的时候再详细介绍其它的flatMap、filter、doOnNext、zip、throttleFirst、forEach等方法吧
3.Observer (观察者)
这里说的观察者不完全是指Observer这个类，观察者是订阅被观察者的事件，并且处理被观察者传递过来的数据的类。
观察者有三个“动作”共同完成完整数据流的处理
public interface Observer<T> {
/**
* Notifies the Observer that the {@link Observable} has finished sending push-based notifications.
* <p>
* The {@link Observable} will not call this method if it calls {@link #onError}.
*/
void onCompleted();
/**
* Notifies the Observer that the {@link Observable} has experienced an error condition.
* <p>
* If the {@link Observable} calls this method, it will not thereafter call {@link #onNext} or
* {@link #onCompleted}.
*
* @param e
* the exception encountered by the Observable
*/
void onError(Throwable e);
/**
* Provides the Observer with a new item to observe.
* <p>
* The {@link Observable} may call this method 0 or more times.
* <p>
* The {@code Observable} will not call this method again after it calls either {@link #onCompleted} or
* {@link #onError}.
*
* @param t
* the item emitted by the Observable
*/
void onNext(T t);
}
使用
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.d(tag, "onCompleted !");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "onError : " + e);
}
@Override
public void onNext(String s) {
Log.d(tag, "onNext : " + s);
}
});
这里看起来是不是很奇怪，像是：一个被观察者注册了一个观察者
NoNoNo，其实这是为了代码的易读和易写，能够很完整的写完一整套响应处理
因为在Rxjava里面的基础动作是Action(无返回参数)和Func(有返回参数)，所以在被观察者Observable中
public final Subscription subscribe(final Observer<? super T> observer) {
...//这里其实将Observer转换成了Subscriber
}
等于
public final Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
...//这里其实将onNext、onError、onComplete转换成了Subscriber
}
因为在Observable中没有直接使用Observer而是最终调用的
private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
...
}
所以最终调用的静态方法，是不是就是像在一个被观察者中订阅一个事件啦。
4.Subscribe (订阅、事件)
是不是很屌，接到上面说的其实：“一个被观察者注册了一个观察者”
其实就是：在一个被观察者中订阅一个事件
那么，subscribe a Subscriber<? super T> to Observable<T> 做了什么呢？下面我用伪代码写一下
private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
// validate and proceed
...
// do nothing by default
subscriber.onStart();
// subscriber是个abstract class当然需要变成实体类啦
if (!(subscriber instanceof SafeSubscriber)) {
//SafeSubscriber就是一个保证了数据传递完整性的实体对象
subscriber = new SafeSubscriber<T>(subscriber);
}
try {
observable.onSubscribe.call(subscriber);
return subscriber;
} catch (Throwable e) {
try {
subscriber.onError(e);
} catch (Throwable e2) {
throw new RuntimeException by e2;
}
//return a static final Unsubscribed Subscription;
return Subscriptions.unsubscribed();
}
}
So , demo可以看上面获取网络数据的伪代码 ↖(ω)↗.
基本讲到这，相信都对RxJava有了个简单的了解。。。。但是
其实上面所有讲的都没有：异步的细节、被观察者的细节、响应式编程思想的分析等等，有空再继续做一些Rxjava造轮子的研究和Android中Rxjava能够带来多大的便利
听说简书还可以打赏支持，有没有老司机带带我 （¯﹃¯）
- 欢迎拍砖，原文链接
- 欢迎交流，QQ：569785819
