rxJava&rxAndroid-基础篇

整理自博客 :http://blog.chengdazhi.com/index.php/140

一、Observable和Observer

RxJava最核心的东西就是Observable和Observer。

Observable 被观察者，会发出数据

Observer 观察者，通过订阅Observable来进行获取Observable的数据。

Observable有2种初始化方式：

方式1：使用Observable.just

Observable<List<String>> listObservable =Observable.just(getColorList());

Observable主动去注册observe：

listObservable.subscribe(new Observer<List<String>>() {

@Override
public void onCompleted() { }

@Override
public void onError(Throwable e) { }

@Override
public void onNext(List<String> colors) {
mSimpleStringAdapter.setStrings(colors);
}
});

一旦Observable通过subscribe()方法订阅Observer之后就会：

onNext()方法被调用，被发送的颜色列表会作为参数传入

不再有数据可以发送，onComplete()方法会被调用。

方式2：使用Observable.fromCallable(···)

Observable<List<String>> tvShowObservable = Observable.fromCallable(new Callable<List<String>>() {

@Override
public List<String> call() {
return mRestClient.getFavoriteTvShows();
}
});

这2者区别是：

使用Observable.fromCallable()，call()中代码（即这里的getFavoriteTvShows（）方法）只会在Observer订阅之后才执行

使用Observable.just(),mRestClient.getFavoriteTvShows()会被立即执行并阻塞UI线程。

-

以下是更复杂的订阅代码：

mTvShowSubscription = tvShowObservable
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<String>>() {

@Override
public void onCompleted() { }

@Override
public void onError(Throwable e) { }

@Override
public void onNext(List<String> tvShows){
displayTvShows(tvShows);
}
});

方法详解：

.subscribeOn(Schedulers.io())：让我们在指定线程中运行获取数据的代码，只要不是UI线程就行。刚才说到的只有在被订阅之后才会执行的代码（上面call() 中的代码），都会在执行subscribe()方法的线程中运行（这里应该不是android UI主线程，但是具体是什么不知道）。而通过subscribeOn()方法，这些代码可以在其他线程中执行。现在即”IO Scheduler”中执行（注意不是android UI主线程）

.observeOn(AndroidSchedulers.mainThread()) : 改变Observer中onNext()方法中执行的线程变为android UI主线程（因为 onNext()涉及到更改View）

mTvShowSubscription = tvShowObservable：前面的mTvShowSubscription就是每当Observer订阅Observable时就会生成一个Subscription对象。一个Subscription代表了一个Observer与Observable之间的连接，我们需要在Activity的onDestroy方法中销毁它:

mTvShowSubscription.unsubscribe();

二、使用Single

Observable还有一个精简版，叫做Single。Single几乎和Observable一模一样，但其回调方法不是onComplete()/onNext()/onError()，而是onSuccess()/onError()。

注意这里Single即可以当Observable也可以当Observer。

Observable代码：

Single<List<String>> tvShowSingle = Single.fromCallable(new Callable<List<String>>() {
@Override
public List<String> call() throws Exception {
mRestClient.getFavoriteTvShows();
}
});

Observer订阅代码：

mTvShowSubscription = tvShowSingle
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleSubscriber<List<String>>() {

@Override
public void onSuccess(List<String> tvShows) {
displayTvShows(tvShows);
}

@Override
public void onError(Throwable error) {
displayErrorMessage();
}
});

订阅一个Single的同时也会自动创建一个Subscription对象。这里的Subscription和案例2中没有区别，一定要在onDestroy()中解除订阅。

三、使用Single和Map

这里把订阅者和被订阅者代码写在一起

Single.just(4).map(new Func1<Integer, String>() {

@Override
public String call(Integer integer) {
return String.valueOf(integer);
}
}).subscribe(new SingleSubscriber<String>() {

@Override
public void onSuccess(String value) {
mValueDisplay.setText(value);
}

@Override
public void onError(Throwable error) { }
});

map(new Func1< Integer, String >)：注意这里前面的Integer参数是输入给Single被订阅者的参数，即这里Single.just(4)中的4，后面的String 接收call()中返回的String类型值，传给订阅者，供 onSuccess(String value) 中使用

四、综合使用

现在我们要写一个用来根据名字搜索城市的Activity。在这个Activity中，我们要使用在这两篇文章中所学的所有知识并写一个比较大的例子。同时还要介绍一个新的概念：deboundce。开始。

现在我们要写一个PublishSubject，并能接收用书输入进输入框的数据，而后根据输入获取符合的列表，并展示。

mTextWatchSubscription = mSearchResultsSubject
.debounce(400, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.io())
.map(new Func1<String, List<String>>() {

@Override
public List<String> call(String s) {
return mRestClient.searchForCity(s);
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<String>>() {

@Override
public void onCompleted() { }

@Override
public void onError(Throwable e) { }

@Override
public void onNext(List<String> cities) {
handleSearchResults(cities);
}
});

mSearchInput.addTextChangedListener(new TextWatcher() {

@Override
public void beforeTextChanged(CharSequence s, int start, int count, int after) { }

@Override
public void onTextChanged(CharSequence s, int start, int before, int count) {
mSearchResultsSubject.onNext(s.toString());
}

@Override
public void afterTextChanged(Editable s) { }
});

这段代码有不少内容，让我们一点一点分析。

首先你会看到debounce()方法。这是啥？有啥用？如果你看一下我们是如何给输入框添加监听器的，你会发现每当输入的内容改变时都会有输入发送到mSearchResultsSubject，不过我们不想让用户每点一个键都向服务器请求一次。我们想等一会，等用户停止输入（代表差不多输完）的时候再请求服务器。

而debounce()方法就是做这个的。这个方法告诉mSearchResultsSubject在没有数据传入达400毫秒时才发送数据。意思就是，仅当用户400ms都没有改变输入内容时，Subject才会发送最新的搜索字符串。这样以来我们就不会进行无意义的网络请求了，UI也不会每输入一个字符都更新。

我们想通过RestClient来访问服务器，而因为RestClient涉及IO操作，我们需要在IO Scheduler中进行这个操作，所以要写observeOn(Schedulers.io())。

好了，现在我们会把搜索字段发送到IO Scheduler中，在这里map就要发挥作用了，我们在map方法中通过关键字获取搜索结果的列表。在map中我们可以调用任意外部方法，在这里使用RestClient获取搜索结果。

因为map方法会在IO Scheduler中运行，而我们又要用其返回值填充View，所以要重新切换到UI线程，所以要写observeOn(AndroidSchedulers.mainThread())。现在搜索结果会被发送到UI线程。要注意两个observeOn()方法的顺序，

注意这里observeOn()方法连续2次调用，注意这里的区别，上面有讲

五、总结与rxJava的优势

这里我们总结一下我们为什么要使用rxJava，之前做过一个项目，是一个利用手机屏幕上发出的光来开锁的项目，里面一些异步的操作就可以使用rxJava来完成