RxAndroid 基础#
参考hi大头鬼hi 的微博，写代码进行测试学习，以下记录共享，同时以便之后查阅。 由于不熟悉lambda，同时开始学习也不建议直接使用lambda，以下大部分代码均使用常规方法编写。
RxJava在github上的地址
RxAndroid在github上的地址
首先，工程中引入：
dependencies {
compile fileTree(dir: 'libs', include: ['*.jar'])
compile 'com.android.support:appcompat-v7:23.0.1'
//引入RxAndroid----begin
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
//引入RxAndroid----end
// RxBinding
compile 'com.jakewharton.rxbinding:rxbinding:0.4.0'
compile 'com.jakewharton.rxbinding:rxbinding-appcompat-v7:0.4.0'
compile 'com.jakewharton.rxbinding:rxbinding-design:0.4.0'
}
一、基本使用
常规语法
Observable<String> myObservable = Observable.create(
new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> sub) {
sub.onNext("Observable Hello, world!");
sub.onCompleted();
}
}
);
Subscriber<String> mySubscriber = new Subscriber<String>() {
@Override
public void onNext(String s) {
L.d("onNext s: " + s);
}
@Override
public void onCompleted() { }
@Override
public void onError(Throwable e) { }
};
myObservable.subscribe(mySubscriber);
简化语法
Observable.just(1)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
L.d("onNext integer: " + integer);
}
});
map操作符变换
Observable.just("Hello, world!")
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s + " -Map";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
L.d("onNext s: " + s);
}
});
二、操作符
Observable.from()
Observable.from()方法，它接收一个集合作为输入，然后每次输出一个元素给subscriber：
List<String> urls = new ArrayList<>();
urls.add("url1");
urls.add("url2");
urls.add("url3");
Observable.from(urls)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
L.d("onNext s: " + s);
}
});
输出如下：
nNext s: url1
onNext s: url2
onNext s: url3
Observable<List<String>> myObservableUrls = Observable.create(
new Observable.OnSubscribe<List<String>>() {
@Override
public void call(Subscriber<? super List<String>> subscriber) {
List<String> urls = new ArrayList<>();
urls.add("url1");
urls.add("url2");
urls.add("url3");
subscriber.onNext(urls);
}
}
);
myObservableUrls.subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
Observable.from(strings)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
L.d("myObservableUrls onNext s: " + s);
}
});
}
});
输出如下：
myObservableUrls onNext s: url1
myObservableUrls onNext s: url2
myObservableUrls onNext s: url3
在前一个Observable输出是list时，嵌套太多，看起来比较乱。
Observable.flatMap()
Observable.flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable。
myObservableUrls.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> urls) {
return Observable.from(urls);
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
L.d("myObservableUrls.flatMap onNext s: " + s);
}
});
避免了上面直接使用Observable.from时需要内如再次subscribe导致的内部嵌套。理解flatMap的关键点在于，flatMap输出的新的Observable正是我们在Subscriber想要接收的。现在Subscriber不再收到List<String>，而是收到一些列单个的字符串，就像Observable.from()的输出一样。
flatMap()，它可以返回任何它想返回的Observable对象。
接着前面的例子，现在我不想打印URL了，而是要打印收到的每个网站的标题。
现在的方法每次只能传入一个URL，并且返回值不是一个String，而是一个输出String的Observabl对象：
private Observable<String> getTitle(final String url) {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("title: " + url);
}
});
}
实现：
myObservableUrls.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> urls) {
return Observable.from(urls);
}
}).flatMap(new Func1<String, Observable<?>>() {
@Override
public Observable<?> call(String s) {
return getTitle(s);
}
}).subscribe(new Action1<Object>() {
@Override
public void call(Object o) {
L.d("myObservableUrls.flatMap onNext o: " + o);
}
});
filter()
filter()输出和输入相同的元素，并且会过滤掉那些不满足检查条件的
myObservableUrls.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> urls) {
return Observable.from(urls);
}
}).flatMap(new Func1<String, Observable<?>>() {
@Override
public Observable<?> call(String s) {
return getTitle(s);
}
}).filter(new Func1<Object, Boolean>() {
@Override
public Boolean call(Object o) {
return !o.equals("title: url2");
}
}).subscribe(new Action1<Object>() {
@Override
public void call(Object o) {
L.d("myObservableUrls.flatMap onNext o: " + o);
}
});
take()
take()输出最多指定数量的结果。
doOnNext()
doOnNext()允许我们在每次输出一个元素之前做一些额外的事情，比如这里的保存标题。
直接上网上的示例:
query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.filter(title -> title != null)
.take(5)
.doOnNext(title -> saveTitle(title))
.subscribe(title -> System.out.println(title));
三、响应式的好处
错误处理
每一个Observerable对象在终结的时候都会调用onCompleted()或者onError()方法，
这种模式有以下几个优点：
1.只要有异常发生onError()一定会被调用
这极大的简化了错误处理。只需要在一个地方处理错误即可以。
2.操作符不需要处理异常
将异常处理交给订阅者来做，Observerable的操作符调用链中一旦有一个抛出了异常，就会直接执行onError()方法。
3.你能够知道什么时候订阅者已经接收了全部的数据。
知道什么时候任务结束能够帮助简化代码的流程。（虽然有可能Observable对象永远不会结束）
使用RXJAVA，OBSERVABLE对象根本不需要知道如何处理错误！操作符也不需要处理错误状态-一旦发生错误，就会跳过当前和后续的操作符。所有的错误处理都交给订阅者来做。
调度器
使用RxJava，你可以使用subscribeOn()指定观察者代码运行的线程，使用observerOn()指定订阅者运行的线程：
myObservableServices.retrieveImage(url)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap));
订阅（Subscriptions）
调用Observable.subscribe()，会返回一个Subscription对象。这个对象代表了被观察者和订阅者之间的联系。
ubscription subscription = Observable.just("Hello, World!")
.subscribe(s -> System.out.println(s));
你可以在后面使用这个Subscription对象来操作被观察者和订阅者之间的联系.
subscription.unsubscribe();
System.out.println("Unsubscribed=" + subscription.isUnsubscribed());
RxJava的另外一个好处就是它处理unsubscribing的时候，会停止整个调用链。如果你使用了一串很复杂的操作符，调用unsubscribe将会在他当前执行的地方终止。不需要做任何额外的工作！
当然也不是所有的代码都使用响应式的方式，仅仅当代码复杂到我想将它分解成简单的逻辑的时候，我才使用响应式代码。
四、在Android中使用响应式编程
这部分主要是摘录，有些方法现在已过时，后续在实际应用中有用到再进行补充更新，
RxAndroid
首先，AndroidSchedulers提供了针对Android的线程系统的调度器。需要在UI线程中运行某些代码？很简单，只需要使用AndroidSchedulers.mainThread():
retrofitService.getImage(url)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap));
throttleFirst(): 在每次事件触发后的一定时间间隔内丢弃新的事件。常用作去抖动过滤，例如按钮的点击监听器：
RxView.clickEvents(button) // RxBinding 代码，后面的文章有解释
.throttleFirst(500, TimeUnit.MILLISECONDS) // 设置防抖间隔为 500ms
.subscribe(subscriber);
遗留代码，运行极慢的代码
绝大多数时候Observable.just() 和 Observable.from() 能够帮助你从遗留代码中创建 Observable 对象:
private Object oldMethod() { ... }
public Observable<Object> newMethod() {
return Observable.just(oldMethod());
}
上面的例子中如果oldMethod()足够快是没有什么问题的，但是如果很慢呢？调用oldMethod()将会阻塞住他所在的线程。
为了解决这个问题，可以参考我一直使用的方法–使用defer()来包装缓慢的代码：
private Object slowBlockingMethod() { ... }
public Observable<Object> newMethod() {
return Observable.defer(() -> Observable.just(slowBlockingMethod()));
}
现在，newMethod()的调用不会阻塞了，除非你订阅返回的observable对象。
生命周期
如何处理Activity的生命周期？主要就是两个问题：
1.在configuration改变（比如转屏）之后继续之前的Subscription。
比如你使用Retrofit发出了一个REST请求，接着想在listview中展示结果。如果在网络请求的时候用户旋转了屏幕怎么办？你当然想继续刚才的请求，但是怎么搞？
2.Observable持有Context导致的内存泄露
这个问题是因为创建subscription的时候，以某种方式持有了context的引用，尤其是当你和view交互的时候，这太容易发生！如果Observable没有及时结束，内存占用就会越来越大。
第一个问题的解决方案就是使用RxJava内置的缓存机制，这样你就可以对同一个Observable对象执行unsubscribe/resubscribe，却不用重复运行得到Observable的代码。cache() (或者 replay())会继续执行网络请求（甚至你调用了unsubscribe也不会停止）。这就是说你可以在Activity重新创建的时候从cache()的返回值中创建一个新的Observable对象。
Observable<Photo> request = service.getUserPhoto(id).cache();
Subscription sub = request.subscribe(photo -> handleUserPhoto(photo));
// ...When the Activity is being recreated...
sub.unsubscribe();
// ...Once the Activity is recreated...
request.subscribe(photo -> handleUserPhoto(photo));
注意，两次sub是使用的同一个缓存的请求。当然在哪里去存储请求的结果还是要你自己来做，和所有其他的生命周期相关的解决方案一延虎，必须在生命周期外的某个地方存储。（retained fragment或者单例等等）。
第二个问题的解决方案就是在生命周期的某个时刻取消订阅。一个很常见的模式就是使用CompositeSubscription来持有所有的Subscriptions，然后在onDestroy()或者onDestroyView()里取消所有的订阅。
private CompositeSubscription mCompositeSubscription
= new CompositeSubscription();
private void doSomething() {
mCompositeSubscription.add(
AndroidObservable.bindActivity(this, Observable.just("Hello, World!"))
.subscribe(s -> System.out.println(s)));
}
@Override
protected void onDestroy() {
super.onDestroy();
mCompositeSubscription.unsubscribe();
}
你可以在Activity/Fragment的基类里创建一个CompositeSubscription对象，在子类中使用它。
注意! 一旦你调用了 CompositeSubscription.unsubscribe()，这个CompositeSubscription对象就不可用了, 如果你还想使用CompositeSubscription，就必须在创建一个新的对象了。
最后，附上测试时使用的完整工程，代码上面均有贴出来了。
RxAndroid基本使用测试代码
http://download.csdn.net/detail/yaodong379/9486905
