rxjava的基础概念
观察者模式于发布-订阅模型
rxjava有四个基础概念
- Observable 被观察对象
- Observer 观察者对象
- subscribe 订阅
- 事件 事件回调，包括onNext、onError、onCompleted
rxjava中事件的概念
-
onCompleted():事件队列完结，不再有onNext()发出时，需要触发onCompleted()方法作为完结标志 -
onError(Throwable e):事件队列异常，同时该事件队列终止，在同一个事件流中，onError和onCompleted只能存在一个 -
onNext(T param):普通的事件回调
rxjava的基本实现
-
创建Observer
直接new一个Observer接口
Observer<String> observer = new Observer<String>() { @Override public void onCompleted() { System.out.println("onCompleted"); } @Override public void onError(Throwable e) { System.out.println(e.getMessage()); } @Override public void onNext(String s) { System,out.println(s); } };或者new一个rxjava中实现了Observer接口的Subscribe抽象类
Subscriber<String> subscriber = new Subscriber<String>() { public void onCompleted() { System.out.println("onCompleted"); } public void onError(Throwable e) { System.out.println(e.getMessage()); } public void onNext(String s) { System.out.println(s + " on onNext"); } }; -
创建Observable对象
Observable<String> observable = Observable.create(subscriber1 -> { System.out.println("call"); subscriber1.onNext("sss"); subscriber1.onNext("sss1"); subscriber1.onCompleted(); }); -
订阅过程，调用subscribe方法
observable.subscribe(subscriber);
通过源码可以发现，subscribe方法最后都会调用
Subscription subscribe(Subscriber<? super T> subscriber)
这个方法，所以第一步在创建观察者对象的时候，即使使用Observer，最终都会转换为Subscriber对象。
这个方法的核心步骤为：
subscriber.onStart(); //可选的准备方法
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber); //Observable被订阅，发出事件流
return Subscriptions.unsubscribed(); //取消订阅
通过查看Subscriber的源码可以发现，它相比Observer多出了2个重要的方法：
- onStart()
在事件发出之前作一些准备工作，默认方法实现为空 - unSubscribe()
取消订阅，释放Observalbe对Subscribe的引用，防止内存泄漏
rxjava的常用操作符
记录一下自己学习的几个rxjava的操作符
-
map
将事件对象转换为另一个对象，例如
Observable.from(strs) .map(new Func1<String, Integer>() { @Override public Integer call(String s) { return s.hashCode(); } }) .subscribe(new Action1<Integer>() { @Override public void call(Integer integer) { System.out.println(integer); } });结合lambda表达式就可以写成为
Observable.from(strs) .map(s -> s.hashCode()) .subscribe(integer -> System.out.println(integer));代码非常的简洁。这段代码获取了每个string的hashcode并进行打印
-
flatMap
先记下这里示例代码中的需要使用对象
static class Student { List<String> courses; //课程 public Student() { } public void setCourses(List<String> courses) { this.courses = courses; } public List<String> getCourses() { return courses; } }有些时候，map并能满足我们的所有需求，比如我使用map，获取到的对象是一个List，这个时候，我们在遍历list的时候就要在回调方法里面写for循环，就类似这样：
Observable.from(students) .map(student -> student.getCourses()).subscribe(strings -> { for (String course : strings) { System.out.println(course); } });如果我需要再把每一个course放入事件流，那么就又需要自己去create Observer，非常的不简洁。于是，我们可以使用flatMap操作符，flatMap操作符可以把事件对象转换为另一个对象的Observable
示例代码如下：Observable.from(students) .flatMap(new Func1<Student, Observable<String>>() { @Override public Observable<String> call(Student student) { return Observable.from(student.getCourses()); } }) .subscribe(new Action1<String>() { @Override public void call(String s) { System.out.println(s); } });配合lamdba写成
Observable.from(students) .flatMap(student -> Observable.from(student.getCourses())) .subscribe(s -> System.out.println(s));即使需要继续下去，也可以继续使用flatMap操作符，代码十分简洁
-
compose操作符
通过源码可以看到，map/flatMap操作符的实现其实是通过left()实现的，在rxjava中，除了left，还有一个变换叫ompose(Transformer)
这个操作符针对自身进行变换，事件序列并不会改变
例如原先的代码中需要多个left()变换:observable1 .lift1() .lift2() .subscribe(subscriber1);
observable2
.lift1()
.lift2()
.subscribe(subscriber2);
observable3
.lift1()
.lift2()
.subscribe(subscriber3);
observable4
.lift1()
.lift2()
.subscribe(subscriber1);
```
这段代码简单封装后可以写成:
```
private Observable liftAll(Observable observable) {
return observable
.lift1()
.lift2()
}
liftAll(observable1).subscribe(subscriber1);
liftAll(observable2).subscribe(subscriber2);
liftAll(observable3).subscribe(subscriber3);
liftAll(observable4).subscribe(subscriber4);
```
这段代码看起来比之前的代码舒服很多，但是封装了observable，感觉不是很灵活，于是代码可以改写成：
```
class AllLeftTransformers implements Observable.Transformer<Integer, String> {
@Override
public Observable<String> call(Observable<Integer> observable) {
return observable.left1().left2();
}
}
AllLeftTransformers liftAll = new AllLiftTransfomers();
//订阅事件
observable1.compose(leftAll).subscriber(subscribe1);
observable2.compose(leftAll).subscriber(subscribe2);
observable3.compose(leftAll).subscriber(subscribe3);
observable4.compose(leftAll).subscriber(subscribe4);
```
这样相对于自己封装的优势就在于，不用把Observable封装，并不会减少Observable的灵活性
rxjava的线程调度
rxjava内置了线程调度(scheduler)
示例代码：
Observable.from(array)
.observeOn(Schedulers.immediate())
.subscribeOn(Schedulers.io())
.subscribe(i -> System.out.println(i + ""));
上面代码的意思就是观察在当前线程，订阅事件发生在io线程
关于rxjava的Schedulers里面常见的线程有如下几种
- Schedulers.immediate() 当前线程
- Schedulers.io() io线程
- Schedulers.newThread() 新的线程
- Schedulers.computation() 计算线程，即CPU密集的线程
其中rxAndroid里面还有一个AndroidSchedulers.mainThread()，表示Android的ui线程
再比如比较经典的，在Android中请求网络数据：
getId()
.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())
.subscriber(user->userView.setId(user.id));
其中
- 观察者表示的就是ui界面，使用在android主线程进行
- 被观察者应该是观察是否发生网络请求获取数据，所以订阅事件发生在io线程2
