功能
关于RxJava的功能，虽说是简化了观察者模式的使用，但是看了网上大家的几个版本的demo，开始还感觉有些疑惑，后来意识到，Rxjava的目的在于简化代码，尤其是引入lamda函数式写法后，更可见一斑。
基于这个目的，也解释了我下面的一个疑惑
- 问：为什么在Observable的subscribe里多次调用onNext?（见下面的代码示例）
- 答： 其实这是示例代码给初接触者带来的歧义。subscribe真实的意图是，在里面顺序的dosomework1, dosomework2，onNext来反馈状态。不要将Emiter引用到subscribe外面来反馈状态 ，不要在subscribe函数体外面处理事务， 理解了这两点，下面的一个问题就好回答了。
- 问： 为什么不给observer和observerable定义局部变量来保存，然后重复利用对象体，而不用每次创建？
- 答： java等高级语言的诞生，首先不是为了简化代码，而是为了更好的呈现流程。即：代码执行效率不是第一考虑因素，代码、框架的可读性才是首要追求。因此 Rxjava在设计之初就不是考虑对象复用，而是使用链式编程理念，尽量简化代码。于是每有一个新的任务，那就创建一各observer和observable去完成就是了，反正它们的代码也比较简洁（虽然最终的.dex文件里的转码不是）。
new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
ALog.i("process and report result");
observableEmitter.onNext("processing1");
observableEmitter.onNext("processing2");
observableEmitter.onComplete();
}
}
简单使用
因为考虑到代码简洁性，所以沿用链式编程，一个sample如下：
Observable.create((ObservableOnSubscribe<String>) observableEmitter -> {
ALog.i("process and report result");
observableEmitter.onNext("processing");
observableEmitter.onComplete();
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable disposable) {
}
@Override
public void onNext(String s) {
ALog.i("observer received:" + s);
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
ALog.i("observer received complete");
}
});
但是这段代码看上去还是比较臃肿，进一步简化如下：
Observable.just("reading").subscribe(s -> {
// show reading in UI thread
ALog.i("observer accept " + s);
});
Observable.just跟Observable.create的效果是一样的，创建一个observable。昨天我还有点不太明白，Observable的subscribe回调函数都省了？我在哪通知呢？
Observable.just的意图很明了，就是在你调用的那个地方的事情处理完了后，直接用Observable.just将状态上报给observer。subscribe函数是多余的，subscribe等价于Observable.just被调的地方所在的函数。
Observable.just有多个参数的使用场合？
看介绍Observable.just可以依次上报多个参数（每个参数的上报等价于使用代码拷贝了一份，虽然隐含的代码量增加了，但是我们的source code还是比较简洁的，所以java考虑的不是效率，而是代码的可读性和功能的独立性，高内聚、低耦合...）
Observable.just使用多参数上报，其实可以简单的考虑为将事务、状态分解，让observer的回调函数一次处理一种状态，而不是 一次顺序的处理多个事务、状态的上报。例如下面的例子：
Observable.just("get up", "eat", "sleep").subscribe(s -> {
if (s.equals("get up")) {
ALog.i("observer accept [thread: " + Thread.currentThread().getName()+ "] display get up");
} else if (s.equals("eat")) {
ALog.i("observer accept [thread: " + Thread.currentThread().getName()+ "] display eat");
} else if (s.equals("sleep")) {
ALog.i("observer accept [thread: " + Thread.currentThread().getName()+ "] display sleep");
}
});
执行结果如下， subscribe的回调，调用了三次：
[Info]:observer accept [thread: main] display get up
[Info]:observer accept [thread: main] display eat
[Info]:observer accept [thread: main] display sleep
可以简单的理解为：代码的caller是被监听的线程，事务完成后，需要将状态转达给监听者，即用RxJava这个调用方式即可。
map的使用
假如有数据列表需要转换数据后再上报，则可以采用map转换操作符。Action0 改名成Action，Action1改名成Consumer，而Action2改名成了BiConsumer，而Action3 - Action9都不再使用了，ActionN变成了Consumer<Object[]>。
代码如下
Observable.just(1, 2, 3, 4, 5)
.map(integer -> {
ALog.i("[thread: " + Thread.currentThread().getName()+ "] " + "worker report " + integer);
return "This is " + integer;
})
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.subscribe(s -> ALog.i("[thread: " + Thread.currentThread().getName()+ "] " + "observer accept " + s));
代码执行情况如下， observer的回调在线程RxNewThreadScheduler-1， 工作者的上报在RxCachedThreadScheduler-1线程。
[Info]:[thread: RxCachedThreadScheduler-1] worker report 1
[Info]:[thread: RxCachedThreadScheduler-1] worker report 2
[Info]:[thread: RxNewThreadScheduler-1] observer accept This is 1
[Info]:[thread: RxNewThreadScheduler-1] observer accept This is 2
[Info]:[thread: RxCachedThreadScheduler-1] worker report 3
[Info]:[thread: RxCachedThreadScheduler-1] worker report 4
[Info]:[thread: RxCachedThreadScheduler-1] worker report 5
[Info]:[thread: RxNewThreadScheduler-1] observer accept This is 3
[Info]:[thread: RxNewThreadScheduler-1] observer accept This is 4
[Info]:[thread: RxNewThreadScheduler-1] observer accept This is 5
