responsive and functional programming RxJava

RxJava因为使用了多个回调，一开始理解起来可能有点难度，其实多看几遍也就明白了，它的招式套路都是同样的：

首先就是建立Observable，建立Observable有不少种方式，这里使用了Observable.create的方式；Observable.create()须要传入一个参数，这个参数实际上是一个回调接口，在这个接口方法里咱们处理开网络请求和解析xml的工做，并在最后经过onNext()、onCompleted()和onError()通知Subscriber（订阅者）;
而后就是调用Observable.subscribe()方法对Observable进行订阅。这里要注意，若是不调用Observable.subscribe()方法，刚才在Observable.create()处理的网络请求和解析xml的代码是不会执行的，这也就解释了本文开头所说的“若是没有观察者(即Subscribers)，Observables是不会发出任何事件的”
说了那么多，好像也没有开线程处理网络请求啊，这样不会报错吗？别急，认真看上面的代码，我还写了两个方法subscribeOn(Schedulers.newThread())和observeOn(AndroidSchedulers.mainThread())，没错，奥妙就在于此：
3.1 subscribeOn(Schedulers.newThread())表示开一个新线程处理Observable.create()方法里的逻辑，也就是处理网络请求和解析xml工做
3.2 observeOn(AndroidSchedulers.mainThread())表示subscriber所运行的线程是在UI线程上，也就是更新控件的操做是在UI线程上
3.3 若是这里只有subscribeOn()方法而没有observeOn()方法，那么Observable.create()和subscriber()都是运行在subscribeOn()所指定的线程中；
3.4 若是这里只有observeOn()方法而没有subscribeOn()方法，那么Observable.create()运行在主线程(UI线程)中，而subscriber()是运行在observeOn()所指定的线程中(本例的observeOn()刚好是指定主线程而已)

Observable<String> myObservable = Observable.create( new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> sub) {
sub.onNext("Hello, world!");
sub.onCompleted(); }
});

Subscriber<String> mySubscriber = new Subscriber<String>() {
@Override public void onNext(String s) { System.out.println(s); }
@Override public void onCompleted() { }
@Override public void onError(Throwable e) { }
};

myObservable.subscribe(mySubscriber);// Outputs "Hello, world!"

public static Observable<WeatherData> getWeatherData(final String city) {
return Observable.create(new Observable.OnSubscribeFunc<WeatherData>() {
@Override
public Subscription onSubscribe(Observer<? super WeatherData> observer) {
try {
observer.onNext(apiManager.getWeather(city, "metric"));
observer.onCompleted();
} catch (Exception e) {
observer.onError(e);
}

return Subscriptions.empty();
}
}).subscribeOn(Schedulers.threadPoolForIO());



public final Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
if (onNext == null) {
throw new IllegalArgumentException("onNext can not be null");
}
if (onError == null) {
throw new IllegalArgumentException("onError can not be null");
}
if (onComplete == null) {
throw new IllegalArgumentException("onComplete can not be null");
}

return subscribe(new Subscriber<T>() {

@Override
public final void onCompleted() {
onComplete.call();
}

@Override
public final void onError(Throwable e) {
onError.call(e);
}

@Override
public final void onNext(T args) {
onNext.call(args);
}

});
}