Rxjava2.0浅析

àì夳堔傛蜴生んèń 2022-09-28 13:22 128阅读 0赞

Rxjava2.0是在Rxjava1的基础上进行了升级  
[Rxjava2.0和Rxjava1的区别][Rxjava2.0_Rxjava1]  
[Rxjava2.0官方Javadoc文档][Rxjava2.0_Javadoc]

## 使用 ##

无论是Rxjava1.0还是2.0，创建Rxjava完整流程分为三步  
**1.创建 Observer**

Observer<WeatherInfoEntity> observer = new Observer<WeatherInfoEntity>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(TAG, "onSubscribe");
                }
    
                @Override
                public void onNext(String value) {
                    Log.d(TAG, value.getCity());
                    weatherView.showWeatherData(value);
                }
    
                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                    Log.d(TAG, "error");
                }
    
                @Override
                public void onComplete() {
    
                }
            };

Rxjava2.0创建的Observer中多了一个回调方法onSubscribe，传递参数为Disposable ，Disposable 相当于RxJava1.x中的Subscription,用于解除订阅。在RxJava1.x中是订阅时返回Disposable，而不是选择回调出来。官方说是为了设计成Reactive-Streams架构。假设Observer需要在接收到异常数据项时解除订阅，在RxJava2.x中则非常简便。只要在onSubscribe中获取Disposable ，然后在onNext（）方法中出现异常数据时，就可以调用disposable.dispose()解除订阅了。

**2.创建 Observable**

Observable<WeatherInfoEntity> observable = Observable.create(new ObservableOnSubscribe<WeatherInfoEntity>() {
               @Override
               public void subscribe(ObservableEmitter<WeatherInfoEntity> e) throws Exception {
                   e.onNext("zoky");
               }
           });

这里跟1.0不同的是：方法的回调的是ObservableEmitter,字面意思即发射器，用于发射数据（onNext）和通知（onError/onComplete）。

**3.建立订阅关系**

observable.subscribe(observer);

这样就完成简单的Rxjava订阅事件流程。

跟Rxjava1一样，Rxjava2跟Retrofit2.x的结合也是听方便的，只要我们在Service中请求方法中返回一个Observable对象，就可以很方便的使用Rxjava来执行网络请求结果的回调。：

public interface WeatherService {
        
        @Headers("Cache-Control: public, max-age=3600")
        @GET("{weatherId}.json")
        Observable<WeatherInfoEntity> getWeatherInfo(@Path("weatherId") String weatherId );
    }

public void getWeatherResult(String areaId) {
            if (areaId == null) {
                Log.d(TAG, "aread is null");
            }
    
     okHttpClient = builder.build();
            Retrofit retrofit = new Retrofit.Builder().client(okHttpClient).baseUrl(BASE_URL)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create()).build();
            service = retrofit.create(WeatherService.class);
    
            service.getWeatherInfo(areaId)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<WeatherInfoEntity>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.d(TAG, "onSubscribe");
                        }
    
                        @Override
                        public void onNext(WeatherInfoEntity value) {
                            Log.d(TAG, value.getCity());
                            weatherView.showWeatherData(value);
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            e.printStackTrace();
                            Log.d(TAG, "error");
                        }
    
                        @Override
                        public void onComplete() {
    
                        }
                    });
    
    
        }

关于操作符，Schedule线程切换原理等，RxJava1.x与RxJava2.x在命名和行为上大多数保持了一致，  
这里推荐看抛物线前辈的： [给Android开发者的Rxjava指南][Android_Rxjava]  
中的操作符部分，我目前能力有限，就不在这里赘述了。虽然现在已经更新到Rxjava2.0了，但这篇文章应该是学习Rxjava的入门经典了。推荐一看~~

[Rxjava2.0_Rxjava1]: https://github.com/ReactiveX/RxJava/wiki/What%27s-different-in-2.0
[Rxjava2.0_Javadoc]: http://reactivex.io/RxJava/2.x/javadoc/
[Android_Rxjava]: http://gank.io/post/560e15be2dca930e00da1083