1、rxjava异步Consumer

导入fxjava包：

implementation 'io.reactivex.rxjava3:rxjava:3.0.6'// rxBinding已经依赖了rxJava，可以无需再重复依赖rxJavaimplementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0'

使用：

getAsync(res -> {System.out.println("返回执行：" + res);});

异步方法：

public void getAsync(Consumer<String> consumer) {new Thread(new Runnable() {@Overridepublic void run() {try {Thread.sleep(5000); // 延时5s} catch (InterruptedException e) {e.printStackTrace();}if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {consumer.accept("返回成功"); // 关键代码}}}).start();}

异步延时5s返回，打印 “返回执行返回成功”

2、合并两个异步都完成才往下运行

public void loadData(){Observable.zip(observableOne(), observableTwo(), new BiFunction<String, String, String>() {@Overridepublic String apply(String a, String b) {return a + b;//合并数据}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String o) {Log.d("TAG==>", o);//两个异步都运行结束才执行这里}});}/*** 异步一* @return*/private Observable observableOne(){Observable observable1 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {SystemClock.sleep(5000);emitter.onNext("我是第一");emitter.onComplete();}}).subscribeOn(Schedulers.io());return observable1;}/*** 异步二* @return*/private Observable observableTwo(){Observable observable2 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {emitter.onNext("我是第二");emitter.onComplete();}}).subscribeOn(Schedulers.io());return observable2;}

3、实现数据缓存优先加载

分别从缓存，硬盘，网络后台获取数据，只要其中一个有数据，后续就停止进行

String memoryCache;String diskCache="我是硬盘数据";public void loadData() {Observable.concat(memoryData(), diskData(), networkData()).firstElement().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {@Overridepublic void accept(String o) {Log.d("获取到的数据为==>", o);}});}/*** 异步一：从缓存获取数据** @return*/private Observable memoryData() {Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) {// 判断是否有数据if (!TextUtils.isEmpty(memoryCache)) {emitter.onNext(memoryCache);} else {emitter.onComplete();}}});return memory;}/*** 异步二：从硬盘获取数据** @return*/private Observable diskData() {Observable<String> disk = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {// 判断是否有数据if (!TextUtils.isEmpty(diskCache)) {emitter.onNext(diskCache);} else {emitter.onComplete();}}});return disk;}/*** 异步三：模仿网络请求** @return*/private Observable networkData() {Observable<String> network = Observable.defer(new Supplier<ObservableSource<? extends String>>() {@Overridepublic ObservableSource<? extends String> get() {SystemClock.sleep(5000);//延时5秒，模仿网络请求return Observable.just("我是网络数据");}});return network;}

4、网络请求失败重试

方式一：

Single.create((SingleOnSubscribe<Integer>) emitter -> {//这里用网络请求emitter.onSuccess(1);//请求成功//emitter.onError(new RuntimeException("请求接口异常..."));//请求失败}).retry((integer, throwable) -> {Log.d("wangyao", "检测是否需要重试:" + integer);//失败一次integer会自动+1return integer < 3;}).subscribe(integer -> {Log.d("wangyao", "请求成功:" + integer);}, throwable -> {Log.d("wangyao", "请求失败:" + throwable);});

方式二：

infoApi.getUserInfo()//总共重试3次，重试间隔3000毫秒.retryWhen(new RetryWithDelay(3, 3000)).observeOn(AndroidSchedulers.mainThread()).subscribeOn(Schedulers.io()).subscribe(new Action1<Response>() {@Overridepublic void call(Response response) {String content = new String(((TypedByteArray) response.getBody()).getBytes());printLog(tvLogs, "", content);}}, new Action1<Throwable>() {@Overridepublic void call(Throwable throwable) {throwable.printStackTrace();}});

android——rxjava异步Consumer类似promise 重试机制 网络请求失败重试 合并两个异步都完成才往下运行 实现数据缓存优先加载