一赏半、概念
Retrofit之所以作為現(xiàn)在最流行的網(wǎng)絡(luò)請(qǐng)求庫(kù)，其中一個(gè)主要原因是支持RxJava屋群，即Retrofit除了提供傳統(tǒng)的網(wǎng)絡(luò)請(qǐng)求方式外开伏，還提供RxJava版本的網(wǎng)絡(luò)請(qǐng)求方式。兩種方式在使用上最大的區(qū)別在于：傳統(tǒng)方式采用了Callback接口冲茸，而RxJava方式則采用了Observable接口屯阀。
主要體現(xiàn)在：
1.用于描述網(wǎng)絡(luò)請(qǐng)求的接口的設(shè)置
// 傳統(tǒng)方式：Call<..>接口形式
public interface GetRequest_Interface {
@GET("url地址")
Call<Translation> getCall();
// 注解里傳入 網(wǎng)絡(luò)請(qǐng)求 的部分URL地址
// getCall()是接受網(wǎng)絡(luò)請(qǐng)求數(shù)據(jù)的方法
}
// RxJava方式：Observable<..>接口形式
@GET("url地址")
Observable<Translation> getCall();
2.網(wǎng)絡(luò)請(qǐng)求的封裝形式和發(fā)送形式
// 傳統(tǒng)方式
// 1. 創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 2. 采用Call<..>接口 對(duì) 發(fā)送請(qǐng)求 進(jìn)行封裝
Call<Translation> call = request.getCall();
// 3. 發(fā)送網(wǎng)絡(luò)請(qǐng)求(異步)
call.enqueue(new Callback<Translation>() {
// 請(qǐng)求成功時(shí)回調(diào)
@Override
public void onResponse(Call<Translation> call, Response<Translation> response) {
...
}
// 請(qǐng)求失敗時(shí)回調(diào)
@Override
public void onFailure(Call<Translation> call, Throwable throwable) {
....
}
});
// RxJava方式
// 1. 創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 2. 采用Observable<...>形式 對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
Observable<Translation> observable = request.getCall();
// 3. 發(fā)送網(wǎng)絡(luò)請(qǐng)求(異步)
observable.subscribeOn(Schedulers.io()) // 在IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 回到主線程 處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
// 發(fā)送請(qǐng)求后調(diào)用該復(fù)寫方法（無(wú)論請(qǐng)求成功與否）
@Override
public void onSubscribe(Disposable d) {
...// 初始化工作
}
// 發(fā)送請(qǐng)求成功后調(diào)用該復(fù)寫方法
@Override
public void onNext(Translation result) {
...// 對(duì)返回結(jié)果Translation類對(duì)象 進(jìn)行處理
}
// 發(fā)送請(qǐng)求成功后，先調(diào)用onNext（）再調(diào)用該復(fù)寫方法
@Override
public void onComplete() {
Log.d(TAG, "請(qǐng)求成功");
}
// 發(fā)送請(qǐng)求失敗后調(diào)用該復(fù)寫方法
@Override
public void onError(Throwable e) {
Log.d(TAG, "請(qǐng)求失敗");
}
});
}
二轴术、使用
1.基本使用
步驟1：添加依賴及網(wǎng)絡(luò)權(quán)限
//添加依賴
// Android 支持 Rxjava
// 此處一定要注意使用RxJava2的版本
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
// Android 支持 Retrofit
compile 'com.squareup.retrofit2:retrofit:2.1.0'
// 銜接 Retrofit & RxJava
// 此處一定要注意使用RxJava2的版本
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
// 支持Gson解析
compile 'com.squareup.retrofit2:converter-gson:2.1.0'
//添加網(wǎng)絡(luò)權(quán)限
<uses-permission android:name="android.permission.INTERNET"/>
步驟2：創(chuàng)建 接收服務(wù)器返回?cái)?shù)據(jù) 的類
public class Translation {
private int status;
private content content;
private static class content {
private String from;
private String to;
private String vendor;
private String out;
private int errNo;
}
//定義 輸出返回?cái)?shù)據(jù) 的方法
public void show() {
System.out.println( "Rxjava翻譯結(jié)果：" + status);
System.out.println("Rxjava翻譯結(jié)果：" + content.from);
System.out.println("Rxjava翻譯結(jié)果：" + content.to);
System.out.println("Rxjava翻譯結(jié)果：" + content.vendor);
System.out.println("Rxjava翻譯結(jié)果：" + content.out);
System.out.println("Rxjava翻譯結(jié)果：" + content.errNo);
}
}
步驟3：創(chuàng)建 用于描述網(wǎng)絡(luò)請(qǐng)求 的接口（區(qū)別于傳統(tǒng)形式）
采用 注解 + Observable<...>接口描述 網(wǎng)絡(luò)請(qǐng)求參數(shù)难衰。
public interface GetRequest_Interface {
@GET("ajax.php?a=fy&f=auto&t=auto&w=hi%20world")
Observable<Translation> getCall();
// 注解里傳入 網(wǎng)絡(luò)請(qǐng)求 的部分URL地址
// Retrofit把網(wǎng)絡(luò)請(qǐng)求的URL分成了兩部分：一部分放在Retrofit對(duì)象里，另一部分放在網(wǎng)絡(luò)請(qǐng)求接口里
// 如果接口里的url是一個(gè)完整的網(wǎng)址逗栽，那么放在Retrofit對(duì)象里的URL可以忽略
// 采用Observable<...>接口
// getCall()是接受網(wǎng)絡(luò)請(qǐng)求數(shù)據(jù)的方法
}
步驟4：創(chuàng)建 Retrofit 實(shí)例
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
步驟5：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
步驟6：對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
采用Observable<...>形式 盖袭。
Observable<Translation> observable = request.getCall();
步驟7：發(fā)送網(wǎng)絡(luò)請(qǐng)求
步驟8：對(duì)返回的數(shù)據(jù)進(jìn)行處理
// 步驟7：發(fā)送網(wǎng)絡(luò)請(qǐng)求
observable.subscribeOn(Schedulers.io()) // 在IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 回到主線程處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始采用subscribe連接");
}
@Override
public void onNext(Translation result) {
// 步驟8：對(duì)返回的數(shù)據(jù)進(jìn)行處理
result.show() ;
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "請(qǐng)求失敗");
}
@Override
public void onComplete() {
Log.d(TAG, "請(qǐng)求成功");
}
});
2.無(wú)條件網(wǎng)絡(luò)請(qǐng)求輪詢
public class RxJavafixRxjava extends AppCompatActivity {
private static final String TAG = "Rxjava";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
/*
* 步驟1：采用interval（）延遲發(fā)送
* 注：此處主要展示無(wú)限次輪詢，若要實(shí)現(xiàn)有限次輪詢彼宠，僅需將interval（）改成intervalRange（）即可
**/
Observable.interval(2,1, TimeUnit.SECONDS)
// 參數(shù)說(shuō)明：
// 參數(shù)1 = 第1次延遲時(shí)間鳄虱；
// 參數(shù)2 = 間隔時(shí)間數(shù)字；
// 參數(shù)3 = 時(shí)間單位凭峡；
// 該例子發(fā)送的事件特點(diǎn)：延遲2s后發(fā)送事件拙已，每隔1秒產(chǎn)生1個(gè)數(shù)字（從0開始遞增1，無(wú)限個(gè)）
/*
* 步驟2：每次發(fā)送數(shù)字前發(fā)送1次網(wǎng)絡(luò)請(qǐng)求（doOnNext（）在執(zhí)行Next事件前調(diào)用）
* 即每隔1秒產(chǎn)生1個(gè)數(shù)字前摧冀，就發(fā)送1次網(wǎng)絡(luò)請(qǐng)求倍踪，從而實(shí)現(xiàn)輪詢需求
**/
.doOnNext(new Consumer<Long>() {
@Override
public void accept(Long integer) throws Exception {
Log.d(TAG, "第 " + integer + " 次輪詢" );
/*
* 步驟3：通過(guò)Retrofit發(fā)送網(wǎng)絡(luò)請(qǐng)求
**/
// a. 創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// b. 創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// c. 采用Observable<...>形式 對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
Observable<Translation> observable = request.getCall();
// d. 通過(guò)線程切換發(fā)送網(wǎng)絡(luò)請(qǐng)求
observable.subscribeOn(Schedulers.io()) // 切換到IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 切換回到主線程 處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Translation result) {
// e.接收服務(wù)器返回的數(shù)據(jù)
result.show() ;
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "請(qǐng)求失敗");
}
@Override
public void onComplete() {
}
});
}
}).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Long value) {
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對(duì)Error事件作出響應(yīng)");
}
@Override
public void onComplete() {
Log.d(TAG, "對(duì)Complete事件作出響應(yīng)");
}
});
}
}
3.有條件網(wǎng)絡(luò)請(qǐng)求輪詢
public class RxJavafixRetrofit extends AppCompatActivity {
private static final String TAG = "Rxjava";
// 設(shè)置變量 = 模擬輪詢服務(wù)器次數(shù)
private int i = 0 ;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 步驟1：創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// 步驟2：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 步驟3：采用Observable<...>形式 對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
Observable<Translation> observable = request.getCall();
// 步驟4：發(fā)送網(wǎng)絡(luò)請(qǐng)求 & 通過(guò)repeatWhen（）進(jìn)行輪詢
observable.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
@Override
// 在Function函數(shù)中，必須對(duì)輸入的 Observable<Object>進(jìn)行處理索昂，此處使用flatMap操作符接收上游的數(shù)據(jù)
public ObservableSource<?> apply(@NonNull Observable<Object> objectObservable) throws Exception {
// 將原始 Observable 停止發(fā)送事件的標(biāo)識(shí)（Complete（） / Error（））轉(zhuǎn)換成1個(gè) Object 類型數(shù)據(jù)傳遞給1個(gè)新被觀察者（Observable）
// 以此決定是否重新訂閱 & 發(fā)送原來(lái)的 Observable建车，即輪詢
// 此處有2種情況：
// 1. 若返回1個(gè)Complete（） / Error（）事件，則不重新訂閱 & 發(fā)送原來(lái)的 Observable椒惨，即輪詢結(jié)束
// 2. 若返回其余事件缤至，則重新訂閱 & 發(fā)送原來(lái)的 Observable，即繼續(xù)輪詢
return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Object throwable) throws Exception {
// 加入判斷條件：當(dāng)輪詢次數(shù) = 5次后康谆，就停止輪詢
if (i > 3) {
// 此處選擇發(fā)送onError事件以結(jié)束輪詢领斥，因?yàn)榭捎|發(fā)下游觀察者的onError（）方法回調(diào)
return Observable.error(new Throwable("輪詢結(jié)束"));
}
// 若輪詢次數(shù)＜4次错洁，則發(fā)送1Next事件以繼續(xù)輪詢
// 注：此處加入了delay操作符，作用 = 延遲一段時(shí)間發(fā)送（此處設(shè)置 = 2s）戒突，以實(shí)現(xiàn)輪詢間間隔設(shè)置
return Observable.just(1).delay(2000, TimeUnit.MILLISECONDS);
}
});
}
}).subscribeOn(Schedulers.io()) // 切換到IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 切換回到主線程 處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Translation result) {
// e.接收服務(wù)器返回的數(shù)據(jù)
result.show() ;
i++;
}
@Override
public void onError(Throwable e) {
// 獲取輪詢結(jié)束信息
Log.d(TAG, e.toString());
}
@Override
public void onComplete() {
}
});
}
}
4.網(wǎng)絡(luò)請(qǐng)求嵌套回調(diào)
未用操作符，代碼如下：
// 發(fā)送注冊(cè)網(wǎng)絡(luò)請(qǐng)求的函數(shù)方法
private void register() {
api.register(new RegisterRequest())
.subscribeOn(Schedulers.io()) //在IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) //回到主線程去處理請(qǐng)求結(jié)果
.subscribe(new Consumer<RegisterResponse>() {
@Override
public void accept(RegisterResponse registerResponse) throws Exception {
Toast.makeText(MainActivity.this, "注冊(cè)成功", Toast.LENGTH_SHORT).show();
login(); //注冊(cè)成功, 調(diào)用登錄的方法
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, "注冊(cè)失敗", Toast.LENGTH_SHORT).show();
}
});
}
// 發(fā)送登錄網(wǎng)絡(luò)請(qǐng)求的函數(shù)方法
private void login() {
api.login(new LoginRequest())
.subscribeOn(Schedulers.io()) //在IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) //回到主線程去處理請(qǐng)求結(jié)果
.subscribe(new Consumer<LoginResponse>() {
@Override
public void accept(LoginResponse loginResponse) throws Exception {
Toast.makeText(MainActivity.this, "登錄成功", Toast.LENGTH_SHORT).show();
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, "登錄失敗", Toast.LENGTH_SHORT).show();
}
});
}
使用flatMap操作符描睦，代碼如下：
public class MainActivity extends AppCompatActivity {
private static final String TAG = "Rxjava";
// 定義Observable接口類型的網(wǎng)絡(luò)請(qǐng)求對(duì)象
Observable<Translation1> observable1;
Observable<Translation2> observable2;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 步驟1：創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// 步驟2：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 步驟3：采用Observable<...>形式 對(duì) 2個(gè)網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
observable1 = request.getCall();
observable2 = request.getCall_2();
observable1.subscribeOn(Schedulers.io()) // （初始被觀察者）切換到IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求1
.observeOn(AndroidSchedulers.mainThread()) // （新觀察者）切換到主線程 處理網(wǎng)絡(luò)請(qǐng)求1的結(jié)果
.doOnNext(new Consumer<Translation1>() {
@Override
public void accept(Translation1 result) throws Exception {
Log.d(TAG, "第1次網(wǎng)絡(luò)請(qǐng)求成功");
result.show();
// 對(duì)第1次網(wǎng)絡(luò)請(qǐng)求返回的結(jié)果進(jìn)行操作 = 顯示翻譯結(jié)果
}
})
.observeOn(Schedulers.io()) // （新被觀察者膊存，同時(shí)也是新觀察者）切換到IO線程去發(fā)起登錄請(qǐng)求
// 特別注意：因?yàn)閒latMap是對(duì)初始被觀察者作變換，所以對(duì)于舊被觀察者忱叭，它是新觀察者隔崎，所以通過(guò)observeOn切換線程
// 但對(duì)于初始觀察者，它則是新的被觀察者
.flatMap(new Function<Translation1, ObservableSource<Translation2>>() { // 作變換韵丑，即作嵌套網(wǎng)絡(luò)請(qǐng)求
@Override
public ObservableSource<Translation2> apply(Translation1 result) throws Exception {
// 將網(wǎng)絡(luò)請(qǐng)求1轉(zhuǎn)換成網(wǎng)絡(luò)請(qǐng)求2爵卒，即發(fā)送網(wǎng)絡(luò)請(qǐng)求2
return observable2;
}
})
.observeOn(AndroidSchedulers.mainThread()) // （初始觀察者）切換到主線程 處理網(wǎng)絡(luò)請(qǐng)求2的結(jié)果
.subscribe(new Consumer<Translation2>() {
@Override
public void accept(Translation2 result) throws Exception {
Log.d(TAG, "第2次網(wǎng)絡(luò)請(qǐng)求成功");
result.show();
// 對(duì)第2次網(wǎng)絡(luò)請(qǐng)求返回的結(jié)果進(jìn)行操作 = 顯示翻譯結(jié)果
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
System.out.println("登錄失敗");
}
});
}
}
5.合并數(shù)據(jù)源 & 同時(shí)展示
使用merge操作符：
// 用于存放最終展示的數(shù)據(jù)
String result = "數(shù)據(jù)源來(lái)自 = " ;
/*
* 設(shè)置第1個(gè)Observable：通過(guò)網(wǎng)絡(luò)獲取數(shù)據(jù)
* 此處僅作網(wǎng)絡(luò)請(qǐng)求的模擬
**/
Observable<String> network = Observable.just("網(wǎng)絡(luò)");
/*
* 設(shè)置第2個(gè)Observable：通過(guò)本地文件獲取數(shù)據(jù)
* 此處僅作本地文件請(qǐng)求的模擬
**/
Observable<String> file = Observable.just("本地文件");
/*
* 通過(guò)merge（）合并事件 & 同時(shí)發(fā)送事件
**/
Observable.merge(network, file)
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String value) {
Log.d(TAG, "數(shù)據(jù)源有： "+ value );
result += value + "+";
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對(duì)Error事件作出響應(yīng)");
}
// 接收合并事件后，統(tǒng)一展示
@Override
public void onComplete() {
Log.d(TAG, "獲取數(shù)據(jù)完成");
Log.d(TAG, result );
}
});
使用zip操作符：
public class MainActivity extends AppCompatActivity {
private static final String TAG = "Rxjava";
// 定義Observable接口類型的網(wǎng)絡(luò)請(qǐng)求對(duì)象
Observable<Translation1> observable1;
Observable<Translation2> observable2;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 步驟1：創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// 步驟2：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 步驟3：采用Observable<...>形式 對(duì) 2個(gè)網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
observable1 = request.getCall().subscribeOn(Schedulers.io()); // 新開線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求1
observable2 = request.getCall_2().subscribeOn(Schedulers.io());// 新開線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求2
// 即2個(gè)網(wǎng)絡(luò)請(qǐng)求異步 & 同時(shí)發(fā)送
// 步驟4：通過(guò)使用zip（）對(duì)兩個(gè)網(wǎng)絡(luò)請(qǐng)求進(jìn)行合并再發(fā)送
Observable.zip(observable1, observable2,
new BiFunction<Translation1, Translation2, String>() {
// 注：創(chuàng)建BiFunction對(duì)象傳入的第3個(gè)參數(shù) = 合并后數(shù)據(jù)的數(shù)據(jù)類型
@Override
public String apply(Translation1 translation1,
Translation2 translation2) throws Exception {
return translation1.show() + " & " +translation2.show();
}
}).observeOn(AndroidSchedulers.mainThread()) // 在主線程接收 & 處理數(shù)據(jù)
.subscribe(new Consumer<String>() {
// 成功返回?cái)?shù)據(jù)時(shí)調(diào)用
@Override
public void accept(String combine_infro) throws Exception {
// 結(jié)合顯示2個(gè)網(wǎng)絡(luò)請(qǐng)求的數(shù)據(jù)結(jié)果
Log.d(TAG, "最終接收到的數(shù)據(jù)是：" + combine_infro);
}
}, new Consumer<Throwable>() {
// 網(wǎng)絡(luò)請(qǐng)求錯(cuò)誤時(shí)調(diào)用
@Override
public void accept(Throwable throwable) throws Exception {
System.out.println("登錄失敗");
}
});
}
}
6.從磁盤 / 內(nèi)存緩存中獲取緩存數(shù)據(jù)
// 該2變量用于模擬內(nèi)存緩存 & 磁盤緩存中的數(shù)據(jù)
String memoryCache = null;
String diskCache = "從磁盤緩存中獲取數(shù)據(jù)";
/*
* 設(shè)置第1個(gè)Observable：檢查內(nèi)存緩存是否有該數(shù)據(jù)的緩存
**/
Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
// 先判斷內(nèi)存緩存有無(wú)數(shù)據(jù)
if (memoryCache != null) {
// 若有該數(shù)據(jù)撵彻，則發(fā)送
emitter.onNext(memoryCache);
} else {
// 若無(wú)該數(shù)據(jù)钓株，則直接發(fā)送結(jié)束事件
emitter.onComplete();
}
}
});
/*
* 設(shè)置第2個(gè)Observable：檢查磁盤緩存是否有該數(shù)據(jù)的緩存
**/
Observable<String> disk = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
// 先判斷磁盤緩存有無(wú)數(shù)據(jù)
if (diskCache != null) {
// 若有該數(shù)據(jù)，則發(fā)送
emitter.onNext(diskCache);
} else {
// 若無(wú)該數(shù)據(jù)陌僵，則直接發(fā)送結(jié)束事件
emitter.onComplete();
}
}
});
/*
* 設(shè)置第3個(gè)Observable：通過(guò)網(wǎng)絡(luò)獲取數(shù)據(jù)
**/
Observable<String> network = Observable.just("從網(wǎng)絡(luò)中獲取數(shù)據(jù)");
// 此處僅作網(wǎng)絡(luò)請(qǐng)求的模擬
/*
* 通過(guò)concat（） 和 firstElement（）操作符實(shí)現(xiàn)緩存功能
**/
// 1. 通過(guò)concat（）合并memory轴合、disk、network 3個(gè)被觀察者的事件（即檢查內(nèi)存緩存碗短、磁盤緩存 & 發(fā)送網(wǎng)絡(luò)請(qǐng)求）
// 并將它們按順序串聯(lián)成隊(duì)列
Observable.concat(memory, disk, network)
// 2. 通過(guò)firstElement()受葛，從串聯(lián)隊(duì)列中取出并發(fā)送第1個(gè)有效事件（Next事件），即依次判斷檢查memory偎谁、disk总滩、network
.firstElement()
// 即本例的邏輯為：
// a. firstElement()取出第1個(gè)事件 = memory，即先判斷內(nèi)存緩存中有無(wú)數(shù)據(jù)緩存巡雨；由于memoryCache = null闰渔，即內(nèi)存緩存中無(wú)數(shù)據(jù)，所以發(fā)送結(jié)束事件（視為無(wú)效事件）
// b. firstElement()繼續(xù)取出第2個(gè)事件 = disk鸯隅，即判斷磁盤緩存中有無(wú)數(shù)據(jù)緩存：由于diskCache ≠ null澜建，即磁盤緩存中有數(shù)據(jù)，所以發(fā)送Next事件（有效事件）
// c. 即firstElement()已發(fā)出第1個(gè)有效事件（disk事件）蝌以，所以停止判斷炕舵。
// 3. 觀察者訂閱
.subscribe(new Consumer<String>() {
@Override
public void accept( String s) throws Exception {
Log.d(TAG,"最終獲取的數(shù)據(jù)來(lái)源 = "+ s);
}
});
7.聯(lián)合判斷多個(gè)事件
/*
* 步驟1：設(shè)置控件變量 & 綁定
**/
EditText name,age,job;
Button list;
name = (EditText) findViewById(R.id.name);
age = (EditText) findViewById(R.id.age);
job = (EditText) findViewById(R.id.job);
list = (Button) findViewById(R.id.list);
/*
* 步驟2：為每個(gè)EditText設(shè)置被觀察者，用于發(fā)送監(jiān)聽事件
* 說(shuō)明：
* 1. 此處采用了RxBinding：RxTextView.textChanges(name) = 對(duì)對(duì)控件數(shù)據(jù)變更進(jìn)行監(jiān)聽（功能類似TextWatcher）跟畅，需要引入依賴：compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
* 2. 傳入EditText控件咽筋，點(diǎn)擊任1個(gè)EditText撰寫時(shí)，都會(huì)發(fā)送數(shù)據(jù)事件 = Function3（）的返回值（下面會(huì)詳細(xì)說(shuō)明）
* 3. 采用skip(1)原因：跳過(guò) 一開始EditText無(wú)任何輸入時(shí)的空值
**/
Observable<CharSequence> nameObservable = RxTextView.textChanges(name).skip(1);
Observable<CharSequence> ageObservable = RxTextView.textChanges(age).skip(1);
Observable<CharSequence> jobObservable = RxTextView.textChanges(job).skip(1);
/*
* 步驟3：通過(guò)combineLatest（）合并事件 & 聯(lián)合判斷
**/
Observable.combineLatest(nameObservable,ageObservable,jobObservable,new Function3<CharSequence, CharSequence, CharSequence,Boolean>() {
@Override
public Boolean apply(@NonNull CharSequence charSequence, @NonNull CharSequence charSequence2, @NonNull CharSequence charSequence3) throws Exception {
/*
* 步驟4：規(guī)定表單信息輸入不能為空
**/
// 1. 姓名信息
boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) ;
// 除了設(shè)置為空徊件，也可設(shè)置長(zhǎng)度限制
// boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) && (name.getText().toString().length() > 2 && name.getText().toString().length() < 9);
// 2. 年齡信息
boolean isUserAgeValid = !TextUtils.isEmpty(age.getText());
// 3. 職業(yè)信息
boolean isUserJobValid = !TextUtils.isEmpty(job.getText()) ;
/*
* 步驟5：返回信息 = 聯(lián)合判斷奸攻，即3個(gè)信息同時(shí)已填寫蒜危，"提交按鈕"才可點(diǎn)擊
**/
return isUserNameValid && isUserAgeValid && isUserJobValid;
}
}).subscribe(new Consumer<Boolean>() {
@Override
public void accept(Boolean s) throws Exception {
/*
* 步驟6：返回結(jié)果 & 設(shè)置按鈕可點(diǎn)擊樣式
**/
Log.e(TAG, "提交按鈕是否可點(diǎn)擊： "+s);
list.setEnabled(s);
}
});
8.線程控制（切換 / 調(diào)度）
public class MainActivity extends AppCompatActivity {
private static final String TAG = "Rxjava";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
//步驟4：創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// 步驟5：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 步驟6：采用Observable<...>形式 對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
Observable<Translation> observable = request.getCall();
// 步驟7：發(fā)送網(wǎng)絡(luò)請(qǐng)求
observable.subscribeOn(Schedulers.io()) // 在IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 回到主線程 處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始采用subscribe連接");
}
@Override
public void onNext(Translation result) {
// 步驟8：對(duì)返回的數(shù)據(jù)進(jìn)行處理
result.show() ;
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "請(qǐng)求失敗");
}
@Override
public void onComplete() {
Log.d(TAG, "請(qǐng)求成功");
}
});
}
}
9.網(wǎng)絡(luò)請(qǐng)求出錯(cuò)重連
public class RxJavafixRetrofit2 extends AppCompatActivity {
private static final String TAG = "RxJava";
// 設(shè)置變量
// 可重試次數(shù)
private int maxConnectCount = 10;
// 當(dāng)前已重試次數(shù)
private int currentRetryCount = 0;
// 重試等待時(shí)間
private int waitRetryTime = 0;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 步驟1：創(chuàng)建Retrofit對(duì)象
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://fy.iciba.com/") // 設(shè)置 網(wǎng)絡(luò)請(qǐng)求 Url
.addConverterFactory(GsonConverterFactory.create()) //設(shè)置使用Gson解析(記得加入依賴)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
.build();
// 步驟2：創(chuàng)建 網(wǎng)絡(luò)請(qǐng)求接口 的實(shí)例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
// 步驟3：采用Observable<...>形式 對(duì) 網(wǎng)絡(luò)請(qǐng)求 進(jìn)行封裝
Observable<Translation> observable = request.getCall();
// 步驟4：發(fā)送網(wǎng)絡(luò)請(qǐng)求 & 通過(guò)retryWhen（）進(jìn)行重試
// 注：主要異常才會(huì)回調(diào)retryWhen（）進(jìn)行重試
observable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
// 參數(shù)Observable<Throwable>中的泛型 = 上游操作符拋出的異常，可通過(guò)該條件來(lái)判斷異常的類型
return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {
// 輸出異常信息
Log.d(TAG, "發(fā)生異常 = "+ throwable.toString());
/**
* 需求1：根據(jù)異常類型選擇是否重試
* 即睹耐，當(dāng)發(fā)生的異常 = 網(wǎng)絡(luò)異常 = IO異常 才選擇重試
*/
if (throwable instanceof IOException){
Log.d(TAG, "屬于IO異常辐赞，需重試" );
/**
* 需求2：限制重試次數(shù)
* 即，當(dāng)已重試次數(shù) < 設(shè)置的重試次數(shù)硝训，才選擇重試
*/
if (currentRetryCount < maxConnectCount){
// 記錄重試次數(shù)
currentRetryCount++;
Log.d(TAG, "重試次數(shù) = " + currentRetryCount);
/**
* 需求2：實(shí)現(xiàn)重試
* 通過(guò)返回的Observable發(fā)送的事件 = Next事件响委，從而使得retryWhen（）重訂閱，最終實(shí)現(xiàn)重試功能
*
* 需求3：延遲1段時(shí)間再重試
* 采用delay操作符 = 延遲一段時(shí)間發(fā)送窖梁，以實(shí)現(xiàn)重試間隔設(shè)置
*
* 需求4：遇到的異常越多赘风，時(shí)間越長(zhǎng)
* 在delay操作符的等待時(shí)間內(nèi)設(shè)置 = 每重試1次，增多延遲重試時(shí)間1s
*/
// 設(shè)置等待時(shí)間
waitRetryTime = 1000 + currentRetryCount* 1000;
Log.d(TAG, "等待時(shí)間 =" + waitRetryTime);
return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);
}else{
// 若重試次數(shù)已 > 設(shè)置重試次數(shù)纵刘，則不重試
// 通過(guò)發(fā)送error來(lái)停止重試（可在觀察者的onError（）中獲取信息）
return Observable.error(new Throwable("重試次數(shù)已超過(guò)設(shè)置次數(shù) = " +currentRetryCount + "邀窃，即 不再重試"));
}
}
// 若發(fā)生的異常不屬于I/O異常，則不重試
// 通過(guò)返回的Observable發(fā)送的事件 = Error事件 實(shí)現(xiàn)（可在觀察者的onError（）中獲取信息）
else{
return Observable.error(new Throwable("發(fā)生了非網(wǎng)絡(luò)異常（非I/O異常）"));
}
}
});
}
}).subscribeOn(Schedulers.io()) // 切換到IO線程進(jìn)行網(wǎng)絡(luò)請(qǐng)求
.observeOn(AndroidSchedulers.mainThread()) // 切換回到主線程 處理請(qǐng)求結(jié)果
.subscribe(new Observer<Translation>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Translation result) {
// 接收服務(wù)器返回的數(shù)據(jù)
Log.d(TAG, "發(fā)送成功");
result.show();
}
@Override
public void onError(Throwable e) {
// 獲取停止重試的信息
Log.d(TAG, e.toString());
}
@Override
public void onComplete() {
}
});
}
}
三假哎、例子
//build.gradle
implementation 'com.squareup.retrofit2:retrofit:2.0.2'
implementation 'com.squareup.retrofit2:converter-gson:2.4.0'
implementation 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
implementation 'io.reactivex.rxjava2:rxjava:2.0.1'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'
//AndroidManifest.xml
<uses-permission android:name="android.permission.INTERNET"/>
//Gist
public class Gist {
Map<String, GistFile> files;
@Override
public String toString() {
StringBuilder stringBuilder = new StringBuilder("[Gist]: ");
if(files != null && files.size() > 0) {
for(Map.Entry entry : files.entrySet()) {
stringBuilder.append(entry.getKey())
.append(":")
.append(entry.getValue())
.append(" ");
}
} else {
stringBuilder.append("null");
}
return stringBuilder.toString();
}
private static class GistFile {
String content;
@Override
public String toString() {
return content;
}
}
}
//HttpService
public interface HttpService {
@GET("gists/c2a7c39532239ff261be")
Observable<Gist> getInfo();
}
//MainActivity
public class MainActivity extends AppCompatActivity {
private static final String TAG = "Retrofit";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Log.d(TAG, "zwm, onCreate");
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("https://api.github.com/")
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
.build();
HttpService httpService = retrofit.create(HttpService.class);
Observable<Gist> observable = httpService.getInfo();
observable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Gist>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "zwm, onSubscribe, thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Gist value) {
Log.d(TAG, "zwm, onNext, thread: " + Thread.currentThread().getName());
Log.d(TAG, "zwm, onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "zwm, onError, thread: " + Thread.currentThread().getName());
}
@Override
public void onComplete() {
Log.d(TAG, "zwm, onComplete, thread: " + Thread.currentThread().getName());
}
});
}
}
//輸出log
2019-12-31 10:20:38.748 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onCreate
2019-12-31 10:20:38.963 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onSubscribe, thread: main
2019-12-31 10:20:41.654 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onNext, thread: main
2019-12-31 10:20:41.656 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onNext: [Gist]: OkHttp.txt:
\\ //
\\ .ooo. //
.@@@@@@@@@.
:@@@@@@@@@@@@@:
:@@. '@@@@@' .@@:
@@@@@@@@@@@@@@@@@
@@@@@@@@@@@@@@@@@
:@@ :@@@@@@@@@@@@@@@@@. @@:
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@ '@@@@@@@@@@@@@@@@@, @@@
@@@@@@@@@@@@@@@@@
'@@@@@@@@@@@@@@@'
@@@@ @@@@
@@@@ @@@@
@@@@ @@@@
'@@' '@@'
:@@@.
.@@@@@@@: +@@ `@@ @@` @@ @@
.@@@@'@@@@: +@@ `@@ @@` @@ @@
@@@ @@@ +@@ `@@ @@` @@ @@
.@@ @@: +@@ @@@ `@@ @@` @@@@@@ @@@@@@ @@;@@@@@
@@@ @@@ +@@ @@@ `@@ @@` @@@@@@ @@@@@@ @@@@@@@@@
@@@ @@@ +@@ @@@ `@@@@@@@@@@` @@ @@ @@@ :@@
@@@ @@@ +@@@@@ `@@@@@@@@@@` @@ @@ @@# @@+
@@@ @@@ +@@@@@+ `@@ @@` @@ @@ @@: @@#
@@: .@@` +@@@+@@ `@@ @@` @@ @@ @@# @@+
@@@. .@@@ +@@ @@@ `@@ @@` @@ @@ @@@ ,@@
@@@@@@@@@ +@@ @@@ `@@ @@` @@@@ @@@@ @@@@#@@@@
@@@@@@@ +@@ #@@ `@@ @@` @@@@: @@@@: @@'@@@@@
@@:
@@:
@@:
2019-12-31 10:20:41.656 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onComplete, thread: main
