一、前言

RxJava 2 是一个流行的 Java 库，用于使用可观察序列组合异步和基于事件的程序。它是原始 RxJava 库的重新实现，旨在更高效并且更适合于 Java 8 及更高版本。

RxJava 2 的主要特性包括：

Observable（可观察对象）：表示随时间推移可以被观察到的数据流或事件。Observable 可以发出零个或多个项目，并且可以成功地终止或以错误终止。

Observer（观察者）：消费 Observable 发出的项目。它通常定义了处理发出的项目、完成事件和错误的方法。

操作符：RxJava 提供了大量的操作符，用于处理、转换和组合 Observable 发出的项目，以便创建更复杂的数据流处理逻辑。

二、安装

// rxjava2
implementation 'com.squareup.retrofit2:adapter-rxjava2:2.9.0'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
implementation 'io.reactivex.rxjava2:rxjava:2.2.20'

三、GET请求实现

接口定义

// rxjava get 请求
@GET("api/users")
// RxUserList 请求函数名，page请求参数
Observable<UserPageParse> RxUserList(@Query("page") Integer page);

逻辑实现

// rxjava2 get 请求
private void Rxjava2GetDemo(){String baseUrl = "https://reqres.in/";Retrofit mRetrofit = new Retrofit.Builder().baseUrl(baseUrl)// 将字符串转换成java对象.addConverterFactory(GsonConverterFactory.create()).addCallAdapterFactory(RxJava2CallAdapterFactory.create()).build();mRetrofit.create(ApiService.class).RxUserList(2).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<UserPageParse>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {tvContent.setText("请求中");}@Overridepublic void onNext(@NonNull UserPageParse userPageParse) {// 处理请求成功的响应Integer page = userPageParse.page;Integer total = userPageParse.total;Integer total_pages = userPageParse.total_pages;// 在这里处理响应数据Log.d(TAG, "异步请求成功" + ',' + page + ',' + total + ',' + total_pages);}@Overridepublic void onError(@NonNull Throwable e) {tvContent.setText("异步请求失败");}@Overridepublic void onComplete() {tvContent.setText("完成");}});
};

四、POST请求实现

接口定义

// rxjava post 请求
@POST("api/users")
// RxUserList 请求函数名，page请求参数
@FormUrlEncoded
Observable<UserParse> RxCreateUser(@Field("name") String name, @Field("job") String job);

逻辑实现

// rxjava2 post 请求
private void Rxjava2PostDemo(){String baseUrl = "https://reqres.in/";Retrofit mRetrofit = new Retrofit.Builder().baseUrl(baseUrl)// 将字符串转换成java对象.addConverterFactory(GsonConverterFactory.create()).addCallAdapterFactory(RxJava2CallAdapterFactory.create()).build();mRetrofit.create(ApiService.class).RxCreateUser("admin", "java").subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<UserParse>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {tvContent.setText("请求中");}@Overridepublic void onNext(@NonNull UserParse UserParse) {// 处理请求成功的响应String name = UserParse.name;String job = UserParse.job;String id = UserParse.id;String createdAt = UserParse.createdAt;// 在这里处理响应数据Log.d(TAG, "异步请求成功" + ',' + name + ',' + job + ',' + id + ',' + createdAt);}@Overridepublic void onError(@NonNull Throwable e) {tvContent.setText("异步请求失败");}@Overridepublic void onComplete() {tvContent.setText("完成");}});
};