RxJava 到底是什么?
异步。
RxJava 的好处?
简洁。随着程序逻辑越来越复杂,它依然保持简洁。
进入主题
基本实现
- 创建Observer
Observer<String> observer = new Observer<String>() {
@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!");
}
};
除了 Observer 接口之外,RxJava 还内置了一个实现了 Observer 的抽象类:Subscriber。 Subscriber 对 Observer 接口进行了一些扩展,但他们的基本使用方式是完全一样的
Subscriber<String> subscriber = new Subscriber<String>() {
@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!");
}
};
例子:
由指定的一个 drawable 文件 id drawableRes 取得图片,并显示在 ImageView 中,并在出现异常的时候打印 Toast 报错:
int drawableRes = ...;
ImageView imageView = ...;
Observable.create(new OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(new Observer<Drawable>() {
@Override
public void onNext(Drawable drawable) {
imageView.setImageDrawable(drawable);
}
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
}
});
2).创建Observable
RxJava 使用 create() 方法来创建一个 Observable ,并为它定义事件触发规则
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
传入了一个 OnSubscribe 对象作为参数。OnSubscribe 会被存储在返回的 Observable 对象中,它的作用相当于一个计划表,当 Observable 被订阅的时候,OnSubscribe 的 call() 方法会自动被调用,事件序列就会依照设定依次触发(对于上面的代码,就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted())。这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递,即观察者模式
RxJava 还提供了一些方法用来快捷创建事件队列,例如:
1、 ————just(T...): 将传入的参数依次发送出来。
Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用:
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
2、————from(T[]) / from(Iterable<? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后,依次发送出来
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用:
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();
3).Subscribe (订阅)
建了 Observable 和 Observer 之后,再用 subscribe() 方法将它们联结起来,整条链子就可以工作了
observable.subscribe(observer);
// 或者:
observable.subscribe(subscriber);
Observable.subscribe(Subscriber) 的内部实现是这样的(仅核心代码):
public Subscription subscribe(Subscriber subscriber) {
subscriber.onStart();
onSubscribe.call(subscriber);
return subscriber;
}
可以看到,subscriber() 做了3件事:
1): 调用 Subscriber.onStart() 。这个方法在前面已经介绍过,是一个可选的准备方法。
2): 调用 Observable 中的 OnSubscribe.call(Subscriber) 。在这里,事件发送的逻辑开始运行。从这也可以看出,在 RxJava 中, Observable 并不是在创建的时候就立即开始发送事件,而是在它被订阅的时候,即当 subscribe() 方法执行的时候。 (订阅的时候才开始发送事件)。
3): 将传入的 Subscriber 作为 Subscription 返回。这是为了方便 unsubscribe().
subscribe() 还支持不完整定义的回调,RxJava 会自动根据定义创建出 Subscriber 。形式如下:
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
Log.d(tag, s);
}
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
// onError()
@Override
public void call(Throwable throwable) {
// Error handling
}
};
Action0 onCompletedAction = new Action0() {
// onCompleted()
@Override
public void call() {
Log.d(tag, "completed");
}
};
// 自动创建 Subscriber ,并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ,并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ,并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
Observer 和 Subscriber 具有相同的角色,而且 Observer 在 subscribe() 过程中最终会被转换成 Subscriber 对象
Action0 是 RxJava 的一个接口,它只有一个方法 call(),这个方法是无参无返回值的;由于 onCompleted() 方法也是无参无返回值的,因此 Action0 可以被当成一个包装对象,将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调,这样其实也可以看做将 onCompleted() 方法作为参数传进了 subscribe()
Action1 也是一个接口,它同样只有一个方法 call(T param),这个方法也无返回值,但有一个参数;与 Action0 同理,由于 onNext(T obj) 和 onError(Throwable error) 也是单参数无返回值的,因此 Action1 可以将 onNext(obj) 和 onError(error) 打包起来传入 subscribe() 以实现不完整定义的回调
例子:
将字符串数组 names 中的所有字符串依次打印出来:
String[] names = ...;
Observable.from(names)
.subscribe(new Action1<String>() {
@Override
public void call(String name) {
Log.d(tag, name);
}
});
分水岭
- 在 RxJava 的默认规则中,事件的发出和消费都是在同一个线程的。 观察者模式本身的目的就是『后台处理,前台回调』的异步机制,因此,则需要用到 RxJava 的另一个概念: Scheduler ,来实现异步。
- 在不指定线程的情况下, RxJava 遵循的是线程不变的原则,即:在哪个线程调用 subscribe(),就在哪个线程生产事件;在哪个线程生产事件,就在哪个线程消费事件。
Scheduler
Scheduler (线程控制)
RxJava 已经内置了几个 Scheduler:
Schedulers.immediate() 默认的,直接在当前线程运行。
Schedulers.newThread() 启用新线程,在新线程工作。
Schedulers.io() I/O操作(读写文件,读写数据库,网络信息交互等)、和newThread()最大的区别是:io() 的内部实现是一个无数量上限的线程池,可以重用空闲的线程。
AndroidSchedulers.mainThread() Android专用的,指定的操作在Android的主线程运行。
使用方法:
subscribeOn() 指定subscribe() 所发生的线程,即 Observable.OnSubcribe 被激活时所处的线程,或者叫事件产生的线程。
observerOn() 指定Subscriber 运行所在的线程,或者叫做事件消费的线程。
文字叙述总归难理解,上代码:
Observable.just(1, 2, 3, 4)
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer number) {
Log.d(tag, "number:" + number);
}
});
上面这段代码中,由于 subscribeOn(Schedulers.io()) 的指定,被创建的事件的内容 1、2、3、4 将会在 IO 线程发出;
而由于observeOn(AndroidScheculers.mainThread()) 的指定,因此subscriber 数字的打印将发生在主线程 。
事实上,这种在subscribe() 之前写上两句subscribeOn(Scheduler.io()) 和observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常见,它适用于多数的 『后台线程取数据,主线程显示』的程序策略。
再看下面这个:
int drawableRes = ...;
ImageView imageView = ...;
Observable.create(new OnSubscribe<Drawable>() {
@Override
public void call(Subscriber<? super Drawable> subscriber) {
Drawable drawable = getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
.subscribe(new Observer<Drawable>() {
@Override
public void onNext(Drawable drawable) {
imageView.setImageDrawable(drawable);
}
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
}
});
那么,加载图片将会发生在 IO 线程,而设置图片则被设定在了主线程。
理解记忆过程
observerOn() 指定的是观察者(警察)的线程
subscribeOn() 指定的是被观察者(小偷)的线程
比如上面的加载图片的例子,在创建警察的时候,newObserver<Drawable>(),就定义好了,警察该干的事,就是setImageDrawable(drawable),设置图片,所以上述例子就是,指定了警察在主线程设置图片。而,在定义小偷的时候,Observable.create,就定了小偷会干什么事,就是获得加载图片的资源,那么上诉例子就是,指定了小偷在IO线程,加载图片.
这样,一抽像一下理解,就加深理解多了。
变换
所谓变换,就是将事件序列中的对象或整个序列进行加工处理,转换成不同的事件或事件序列
map
例子1
Observable.just("images/logo.png") // 输入类型 String
.map(new Func1<String, Bitmap>() {
@Override
public Bitmap call(String filePath) { // 参数类型 String
return getBitmapFromPath(filePath); // 返回类型 Bitmap
}
})
.subscribe(new Action1<Bitmap>() {
@Override
public void call(Bitmap bitmap) { // 参数类型 Bitmap
showBitmap(bitmap);
}
});
Func1,它和 Action1 非常相似,也是 RxJava 的一个接口,用于包装含有一个参数的方法。 Func1 和 Action 的区别在于, Func1 包装的是有返回值的方法。
可以看到,map() 方法将参数中的 String 对象转换成一个 Bitmap 对象后返回,而在经过 map() 方法后,事件的参数类型也由 String 转为了 Bitmap。
它不仅可以针对事件对象,还可以针对整个事件队列。
flatMap
首先假设这么一种需求:假设有一个数据结构『学生』,现在需要打印出一组学生的名字。实现方式很简单
private Students s1 = new Students("赵日天");
private Students s2 = new Students("陈铁柱");
private Students s3 = new Students("王尼玛");
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
list.add(s1);
list.add(s2);
list.add(s3);
Observable.from(list).map(new Func1<Students, String>() {
@Override
public String call(Students stu) {
// TODO Auto-generated method stub
return stu.getName();
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println("Students name is :" + s);
}
});
}
class Students {
private String name;
public Students(String name) {
this.name = name;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
运行结果:
那么再假设:如果要打印出每个学生所需要修的所有课程的名称呢?(需求的区别在于,每个学生只有一个名字,但却有多个课程。)首先可以这样实现
private List<Course> courseList = new ArrayList<MainActivity.Course>();
private Course c1 = new Course("紅Buff");
private Course c2 = new Course("蓝Buff");
private Course c3 = new Course("火龙");
private Students s1;
private Students s2;
private Students s3;
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
courseList.add(c1);
courseList.add(c2);
courseList.add(c3);
s1 = new Students("赵日天", courseList);
s2 = new Students("陈铁柱", courseList);
s3 = new Students("王尼玛", courseList);
list.add(s1);
list.add(s2);
list.add(s3);
Observable.from(list).subscribe(new Subscriber<Students>() {
@Override
public void onCompleted() {
// TODO Auto-generated method stub
}
@Override
public void onError(Throwable arg0) {
// TODO Auto-generated method stub
}
@Override
public void onNext(Students student) {
// TODO Auto-generated method stub
List<Course> courses = student.getCourseList();
for (int i = 0; i < courses.size(); i++) {
Course course = courses.get(i);
System.out.println("Student " + student.getName()
+ "Course name is :" + course.getcourseName());
}
}
});
}
// 模拟学生结构
class Students {
private String name;
private List<Course> courseList;
public Students(String name, List<Course> courseList) {
this.name = name;
this.courseList = courseList;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public List<Course> getCourseList() {
return courseList;
}
public void setCourseList(List<Course> courseList) {
this.courseList = courseList;
}
}
// 模拟课程数据结构
class Course {
private String courseName;
public Course(String courseName) {
this.courseName = courseName;
}
public String getcourseName() {
return courseName;
}
public void setcourseName(String name) {
this.courseName = name;
}
}
运行结果:
代码很简单。但是,如果不想在 Subscriber 中使用 for 循环呢,而是希望 Subscriber 中直接传入单个的 Course 对象呢?因为map() 是一对一的转化,而现在的要求是一对多的转化。那怎么才能把一个 Student 转化成多个 Course 呢?
这个时候,就需要用 flatMap() 了
flatMap
private List<Course> courseList = new ArrayList<MainActivity.Course>();
private Course c1 = new Course("紅Buff");
private Course c2 = new Course("蓝Buff");
private Course c3 = new Course("火龙");
private Students s1;
private Students s2;
private Students s3;
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
courseList.add(c1);
courseList.add(c2);
courseList.add(c3);
s1 = new Students("赵日天", courseList);
s2 = new Students("陈铁柱", courseList);
s3 = new Students("王尼玛", courseList);
list.add(s1);
list.add(s2);
list.add(s3);
Observable.from(list)
.flatMap(new Func1<Students, Observable<Course>>() {
@Override
public Observable<Course> call(Students stu) {
// TODO Auto-generated method stub
System.out.println("学生" + stu.getName() + "的课程如下:");
return Observable.from(stu.getCourseList());
}
}).subscribe(new Action1<Course>() {
@Override
public void call(Course cou) {
System.out.println(cou.getcourseName());
}
});
}
// 模拟学生结构
class Students {
private String name;
private List<Course> courseList;
public Students(String name, List<Course> courseList) {
this.name = name;
this.courseList = courseList;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public List<Course> getCourseList() {
return courseList;
}
public void setCourseList(List<Course> courseList) {
this.courseList = courseList;
}
}
// 模拟课程数据结构
class Course {
private String courseName;
public Course(String courseName) {
this.courseName = courseName;
}
public String getcourseName() {
return courseName;
}
public void setcourseName(String name) {
this.courseName = name;
}
}
运行结果:
可以看出, flatMap() 和 map() 有一个相同点:它也是把传入的参数转化之后返回另一个对象。但需要注意,和 map() 不同的是, flatMap() 中返回的是个 Observable 对象,并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。
flatMap() 的原理
- 使用传入的事件对象创建一个 Observable 对象;
- 并不发送这个 Observable, 而是将它激活,于是它开始发送事件;
- 每一个创建出来的 Observable 发送的事件,都被汇入同一个 Observable,而这个 Observable 负责将这些事件统一交给 Subscriber 的回调方法。
个人理解
特工(被观察者)的能力是给他传入一堆学生,他先自己查到这个学生的全部课程,由于每个学生的课程都很多,所以不能直接让监视他的上级知道,他自己有一个情报小组,把课程给到情报小组,然后整理好情报,最后由情报小组统一再给到时刻关注监视他的上级(观察者)。在该事件中,特工把事件拆成了两级,通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。也就是,把每一个学生的每一项课程,铺平后,按顺序的列好告诉上级。
说点什么吧
终于到这了,说句实话,扔物线的那篇文章是真的棒,我看了三遍才能看得懂,第一遍看的时候,整个人都是懵逼状态的,天啊,这什么啊可是,必须硬着头皮看下去啊,看第二遍的时候,就感觉很舒服了,看到运行结果的时候,真的棒棒大。(⊙o⊙)…
看完这个,结合了基于Retrofit/RxJava/mvp+..来一发Demo先:
https://github.com/GitHuborz/MvpDemo-RetrofitRxJvaDemo
该工程包括 :
1、MvpDemo (一个简单的MVP架构的demo)
2、RetrofitRxJavaDemo (一个使用RxJava + Retrofit 的小demo)
MVP是看鸿洋大大的这个,理解造轮子的。 :http://blog.csdn.net/lmj623565791/article/details/46596109
至于RxJava 和 Retrofit ,看的是抛物线的文章,还有别的一些。这里有Fork了一个挺好的搜集,不过,个人认为,有很多选择看就好了,有些不一定好。 :https://github.com/GitHuborz/Awesome-RxJava
最近的计划是,学完这些后,写一个用Gank.io的接口,基于 RxJava+Retrofit+MVP+OkHttp,当然也要使用时下流行许久了的Rv+MD等等的一个妹子APP。(哈哈)
总之,一步一步实现吧,Go~
想看妹子,请看下回分解~