RxJava Android初探和常用实战代码

下面的代码是基于RX1.0

//创建被观察者 也就是事件发布者 Subscriber 因此 SubscriberOn实际上是控制call回调应该所处线程
Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {@Overridepublic void call(Subscriber<? super Integer> observer) {//我的回调在哪个线程由 .subscribeOn(AndroidSchedulers.mainThread()  发布者在哪个线程方法控制的Log.w(TAG, "onCall" + Thread.currentThread().getName());for (int i = 0; i < 5; i++) {observer.onNext(i);}observer.onCompleted();}
});//不订阅  call就不会触发
//观察回调在io线程 订阅回调在主线程
observable.observeOn(Schedulers.io()).subscribeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Integer>() {@Overridepublic void onCompleted() {Log.w(TAG, "Observable completed");}@Overridepublic void onError(Throwable e) {Log.w(TAG, "Oh,no! Something wrong happened！");}@Overridepublic void onNext(Integer item) {//这里的观察方法回调在哪个线程由observeOn 控制的 另外 不订阅 就不会产生回调Log.w(TAG, "Item is " + item + ",Thread:" + Thread.currentThread().getName());}
});如果不写 observeOn 和 subscribeOn则默认在调用者所在线程，s W/Example1Activity: onCallRxCachedThreadScheduler-1
10-24 16:42:04.960 8526-8613/kurtis.rx.androidexamples W/Example1Activity: Item is 0,Thread:RxCachedThreadScheduler-2

例子2 简单的适配器 与耗时请求更新模板代码

private void createObservable() {Observable<List<String>> tvShowObservable = Observable.fromCallable(new Callable<List<String>>() {@Overridepublic List<String> call() {Log.w(TAG, "我当前所在线程非ui线程" + Thread.currentThread().getName());return mRestClient.getFavoriteTvShows();//这里调用了sleep 因为使用了 subscribeOn}});mTvShowSubscription = tvShowObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<List<String>>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(List<String> tvShows) {//当onCall完成之后将调用这个。displayTvShows(tvShows);}});}@Override
protected void onDestroy() {super.onDestroy();if (mTvShowSubscription != null && !mTvShowSubscription.isUnsubscribed()) {mTvShowSubscription.unsubscribe();}
}

错误与成功回调的模板代码

private void createSingle() {Single<List<String>> tvShowSingle = Single.fromCallable(new Callable<List<String>>() {@Overridepublic List<String> call() throws Exception {/*** Uncomment me (and comment out the line below) to see what happens when an error occurs.** return RestClient.getFavoriteTvShowsWithException();*/return mRestClient.getFavoriteTvShows();// 使用 getFavoriteTvShowsWithException方法 由于抛出了异常 所以会毁掉onError}});mTvShowSubscription = tvShowSingle.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new SingleSubscriber<List<String>>() {@Overridepublic void onSuccess(List<String> tvShows) {displayTvShows(tvShows);}@Overridepublic void onError(Throwable error) {displayErrorMessage();}});
}

需要手动调用才会触发回调的 家伙

private void createCounterEmitter() {mCounterEmitter = PublishSubject.create();mCounterEmitter.subscribe(new Observer<Integer>() {@Overridepublic void onCompleted() {Log.w(TAG,"onCompleted:");}@Overridepublic void onError(Throwable e) {Log.w(TAG,"onError:"+e);}@Overridepublic void onNext(Integer integer) {Log.w(TAG,"onNext:"+integer);mCounterDisplay.setText(String.valueOf(integer));}});
}private void onIncrementButtonClick() {mCounter++;Log.w(TAG,"request call onNext:"+mCounter);mCounterEmitter.onNext(mCounter);}

参考链接

http://www.jianshu.com/p/669eda5dc5a4http://blog.csdn.net/chen_zhang_yu/article/details/52900725