整理自博客 :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之后就会：

1. onNext()方法被调用，被发送的颜色列表会作为参数传入
2. 不再有数据可以发送，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线程。

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

（tvShowObservable即上面创建的Observable对象）

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) 中使用（注意这里用到了map，将Integer变换成了String，这个变换过程会在进阶篇详细讲解）

四、综合使用

现在我们要写一个用来根据名字搜索城市的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来完成

···