RxJava 的使用入门

傷城~ 2023-10-18 12:41 25阅读 0赞

一、什么是 Rxjava？  
RxJava 是一个响应式编程框架，采用观察者设计模式。所以自然少不了 Observable 和 Subscriber 这两个东东了。  
RxJava 是一个开源项目，地址：[https://github.com/ReactiveX/RxJava][https_github.com_ReactiveX_RxJava]

还有一个RxAndroid，用于 Android 开发，添加了 Android 用的接口。地址：[https://github.com/ReactiveX/RxAndroid][https_github.com_ReactiveX_RxAndroid]  
二、例子  
通过请求openweathermap 的天气查询接口返回天气数据  
1、增加编译依赖

1 dependencies {
    2     compile fileTree(dir: "libs", include: ["*.jar"])
    3     compile "com.android.support:appcompat-v7:22.0.0"
    4     compile "io.reactivex:rxjava:1.0.9"
    5     compile "io.reactivex:rxandroid:0.24.0"
    6     compile "com.squareup.retrofit:retrofit:1.9.0"
    7 }
    
    
    retrofit 是一个 restful 请求客户端。详见：http://square.github.io/retrofit/
    2、服务器接口
    
    1 /**
    2  * 接口
    3  * Created by Hal on 15/4/26.
    4  */
    5 public class ApiManager {
    6
    7     private static final String ENDPOINT = "http://api.openweathermap.org/data/2.5";
    8
    9     /**
    10      * 服务接口
    11      */
    12     private interface ApiManagerService {
    13         @GET("/weather")
    14         WeatherData getWeather(@Query("q") String place, @Query("units") String units);
    15     }
    16
    17     private static final RestAdapter restAdapter = new RestAdapter.Builder().setEndpoint(ENDPOINT).setLogLevel(RestAdapter.LogLevel.FULL).build();
    18
    19     private static final ApiManagerService apiManager = restAdapter.create(ApiManagerService.class);
    20
    21     /**
    22      * 将服务接口返回的数据，封装成{@link rx.Observable}
    23      * @param city
    24      * @return
    25      */
    26     public static Observable<WeatherData> getWeatherData(final String city) {
    27         return Observable.create(new Observable.OnSubscribe<WeatherData>() {
    28             @Override
    29             public void call(Subscriber<? super WeatherData> subscriber) {
    30                 //订阅者回调 onNext 和 onCompleted
    31                 subscriber.onNext(apiManager.getWeather(city, "metric"));
    32                 subscriber.onCompleted();
    33             }
    34         }).subscribeOn(Schedulers.io());
    35     }
    36 }

订阅者的回调有三个方法，onNext，onError，onCompleted  
3、接口调用

1   /**
    2          * 多个 city 请求
    3          * map，flatMap 对 Observable进行变换
    4          */
    5         Observable.from(CITIES).flatMap(new Func1<String, Observable<WeatherData>>() {
    6             @Override
    7             public Observable<WeatherData> call(String s) {
    8                 return ApiManager.getWeatherData(s);
    9             }
    10         }).subscribeOn(Schedulers.io())
    11                 .observeOn(AndroidSchedulers.mainThread())
    12                 .subscribe(/*onNext*/new Action1<WeatherData>() {
    13                     @Override
    14                     public void call(WeatherData weatherData) {
    15                         Log.d(LOG_TAG, weatherData.toString());
    16                     }
    17                 }, /*onError*/new Action1<Throwable>() {
    18                     @Override
    19                     public void call(Throwable throwable) {
    20
    21                     }
    22                 });
    23
    24         /**
    25          * 单个 city 请求
    26          */
    27         ApiManager.getWeatherData(CITIES[0]).subscribeOn(Schedulers.io())
    28                 .observeOn(AndroidSchedulers.mainThread())
    29                 .subscribe(new Action1<WeatherData>() {
    30                     @Override
    31                     public void call(WeatherData weatherData) {
    32                         Log.d(LOG_TAG, weatherData.toString());
    33                         ((TextView) findViewById(R.id.text)).setText(weatherData.toString());
    34                     }
    35                 }, new Action1<Throwable>() {
    36                     @Override
    37                     public void call(Throwable throwable) {
    38                         Log.e(LOG_TAG, throwable.getMessage(), throwable);
    39                     }
    40                 });
    41
    42         /**
    43          * Android View 事件处理
    44          */
    45         ViewObservable.clicks(findViewById(R.id.text), false).subscribe(new Action1<OnClickEvent>() {
    46             @Override
    47             public void call(OnClickEvent onClickEvent) {
    48
    49             }
    50         });

subscribeOn([Schedulers.io][]())与observeOn(AndroidSchedulers.mainThread())分别定义了这两个动作的线程。Android UI 更新需要在主线程。  
4、retrofit 支持 rxjava 整合

1 /**
    2      * 服务接口
    3      */
    4     private interface ApiManagerService {
    5         @GET("/weather")
    6         WeatherData getWeather(@Query("q") String place, @Query("units") String units);
    7
    8         /**
    9          * retrofit 支持 rxjava 整合
    10          * 这种方法适用于新接口
    11          */
    12         @GET("/weather")
    13         Observable<WeatherData> getWeatherData(@Query("q") String place, @Query("units") String units);
    14     }

[https_github.com_ReactiveX_RxJava]: https://github.com/ReactiveX/RxJava
[https_github.com_ReactiveX_RxAndroid]: https://github.com/ReactiveX/RxAndroid
[Schedulers.io]: http://Schedulers.io