前言
之前在github开源过一个网络库,这是一款基于RxJava2+Retrofit2实现简单易用的网络请求框架。在这里对网络库的用法就不做过多介绍,感兴趣的可以去了解下。在使用过程中一些网友反馈不知道怎么结合Rxjava2来实现一些场景需求,希望能够写一遍文章介绍下。终于抽出时间来对与Rxjava2在实际场景使用的一些案例做个简单的总结和介绍。不知道怎么使用,主要是对RxEasyHttp有个误区,RxEasyHttp不仅是支持采用链式调用一点到底方便使用,同时也支持返回Observable
的用法,拿到了Observable
自然就可以很好的利用Rxjava操作符了来实现各种强大的功能。本文主要是讲RxEasyHttp与Rxjava2怎么结合的,也不会Rxjava2的操作符深入讲解,不然就脱离了本文的重心!废话不多说了,一起来看看是如何使用的。
场景介绍
场景一:延迟请求
在页面网络接口请求中,不是希望立马请求,而是需要延迟指定的时间后再去请求。
延迟请求:利用RxJava的timer
操作符。timer:主要作用就是创建一个Observable
,它在一个给定的延迟后发射一个特殊的值,只是延迟发送一次并不会按照周期执行。timer()
源码如下: public static Observabletimer(long delay, TimeUnit unit) { return timer(delay, unit, Schedulers.computation()); }
可以看到采用timer()
返回的是Observable<Long>
,而网络请求返回的Observable
并不是Observable<Long>
,如何将这两个Observable
关联起来,就需要采用另外一个操作符flatMap()
,简单理解就是flatMap
使用一个指定的函数对原始Observable发射的每一项数据进行相应的变换操作。flatMap详细作用不做过多介绍。
//延迟5s请求 Observable.timer(5, TimeUnit.SECONDS).flatMap(new Function>() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { //延迟结束开始执行网络请求 Observable observable = EasyHttp.get("/v1/app/chairdressing/skinAnalyzePower/skinTestResult") .timeStamp(true) .execute(SkinTestResult.class); return observable; } }).subscribe(new BaseSubscriber () { @Override protected void onStart() { } @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull SkinTestResult skinTestResult) { Log.i("test", "=====" + skinTestResult.toString()); } }); //在不需要轮询的时候,取消轮询 //EasyHttp.cancelSubscription(polldisposable);
- timer在这里作用延迟5s结束时就会触发网络请求
- flatMap在这里的作用就是将
timer
操作符返回的Observable<Long>
和网络请求的Observable<SkinTestResult>
做转换,在subscribe订阅时返回的内容,我们真正需要的SkinTestResult,而不是Long. 因此将Observable<Long>
变换成Observable<SkinTestResult>
输出SkinTestResult,完美达到目的。
场景二:轮询请求-无限轮询
在项目中需要用到每隔5s刷新一次页面或者拉取最新消息。轮询器大家一定不陌生,开发中无论是Java的Timer+TimeTask , 还是Android的Hanlder都可实现,现在介绍另一种简单的实现方式。
无限轮询:利用RxJava的Interval
操作符。interval:创建一个按固定时间间隔发射整数序列的Observable,它是按照周期执行的。源码如下(只展示相关的两个方法): public static Observableinterval(long initialDelay, long period, TimeUnit unit) { return interval(initialDelay, period, unit, Schedulers.computation()); } public static Observable interval(long period, TimeUnit unit) { return interval(period, period, unit, Schedulers.computation()); }
可以看到采用interval()
返回的是Observable<Long>
,而网络请求返回的Observable
并不是Observable<Long>
,如何将这两个Observable
关联起来,就需要采用另外一个操作符flatMap()
,简单理解就是flatMap使用一个指定的函数对原始Observable发射的每一项数据之行相应的变换操作。flatMap详细作用不做过多介绍(同上场景一)。
//自己根据需要选择合适的interval方法 Disposable polldisposable = Observable.interval(0, 5, TimeUnit.SECONDS).flatMap(new Function>() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { return EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); } }).subscribeWith(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull Content content) { showToast(content.toString()); } }); //在不需要轮询的时候,取消轮询 //EasyHttp.cancelSubscription(polldisposable);
- interval在这里作用每隔5s结束时就会触发网络请求
- 注意
interval(0, 5, TimeUnit.SECONDS)
和interval(5, TimeUnit.SECONDS)
的区别,自己根据需要选择合适的interval方法。
interval(0,5, TimeUnit.SECONDS)
:3个参数,第一个参数表示初始化延时多久开始请求,这里用0表示不延时直接请求,第二个参数表示间隔多久轮询一次,这里表示间隔5s,第三个表示设置的时间单位。interval(5, TimeUnit.SECONDS)
:2个参数,其中的这个5就表示,初始延时5秒开始执行请求,轮询也是5s,第二个表示设置的时间单位,从上面提供的interval()
源码可以看出。
flatMap
在这里的作用就是将interval
的Observable<Long>
和网络请求的Observable<Content>
做转换,输出Content,而不是Long.
场景三:轮询请求-限定次数轮询
这个和无限轮询用法基本一样,只是多了轮询的次数限制条件,不是一直无限的轮询下去。
轮询次数:利用RxJava的intervalRange
或者take
操作符。intervalRange:以一个例子说明可能更清楚,intervalRange(0,3,0,5, TimeUnit.SECONDS)
表示从0开始输出3个数据,延迟0秒执行,每隔5秒执行一次。take:表示只取前n项。这里用take和interval操作符联合使用,由于一旦interval计时开始除了解绑就无法停止,使用take操作符就简单很多了,它的意思是只释放前n项,过后Observable流就自动终止。例如:只轮询3次 int count = 3;//轮询3次 //方式一:采用intervalRange //Observable.intervalRange(0,count,0,5, TimeUnit.SECONDS).flatMap(new Function>() { //方式一:采用take countdisposable = Observable.interval(0, 5, TimeUnit.SECONDS).take(count).flatMap(new Function >() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { return EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); } }).subscribeWith(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull Content content) { showToast(content.toString()); } }); //在不需要轮询的时候,取消轮询 //EasyHttp.cancelSubscription(polldisposable);
场景四:轮询请求-条件轮询
条件轮询和限定次数轮询比较像,都是起达到目的后终止轮询。比如一个网络请求一直在轮询执行,直到获取到了想要的内容后就终止掉轮询。
条件轮询:利用RxJava的takeUntil
操作符。takeUntil:使用一个标志Observable是否发射数据来判断,当标志Observable没有发射数据时,正常发射数据,而一旦标志Observable发射过了数据则后面的数据都会被丢弃。例如:轮询请求中如果返回的内容字符串中包含“示”就终止轮询 Observable.interval(0, 5, TimeUnit.SECONDS).flatMap(new Function>() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { return EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); } }).takeUntil(new Predicate () { @Override public boolean test(@NonNull Content content) throws Exception { //如果条件满足,就会终止轮询,这里逻辑可以自己写 //结果为true,说明满足条件了,就不在轮询了 return content.getOut().contains("示"); } }).subscribeWith(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull Content content) { showToast(content.toString()); } }); //在不需要轮询的时候,取消轮询 //EasyHttp.cancelSubscription(polldisposable);
场景五:轮询请求-过滤轮询
过滤轮询主要是指在轮询的过程中对订阅的内容做过滤,不是需要的内容就不会返回给订阅者,但是它不会中断轮询。过滤轮询也可以理解成是无限轮询加了一个过滤条件而已。
过滤轮询:利用Rxjava的filter
操作符。filter:是对源Observable产生的结果按照指定条件进行过滤,只有满足条件的结果才会提交给订阅者。例如:返回的状态码如果是错误就不返回给订阅者,不更新界面(只有保证每次请求成功才刷新界面),但是会继续轮询请求 Disposable filterdisposable = Observable.interval(0, 5, TimeUnit.SECONDS).flatMap(new Function>() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { return EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); } }).filter(new Predicate () { @Override public boolean test(@NonNull Content content) throws Exception { //如果不满足条件就过滤该条轮询数据,但是轮询还是一直执行 //ErrNo==0表示成功,如果不等于0就认为失败,content不会返回给订阅者 return content.getErrNo() != 0; } }).subscribeWith(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull Content content) { showToast(content.toString()); } }); //在不需要轮询的时候,取消轮询 //EasyHttp.cancelSubscription(polldisposable);
filter
操作符在这里只是为了举列说明,是自己为了讲解定义了一个过滤轮询的概念,不是说filter
只能在轮询这里使用,它是可以和其它任何Rxjava操作符配合使用。- 切记
takeUntil
和filter
的区别,takeUntil
找到自己想要的数据后就结束了流,不再执行任何操作。filter
发现不符合条件的不会给订阅者,只有符合条件的才给订阅者,发现不符合的,不会中断操作。
场景六:嵌套请求
在开发中由于请求网络数据频繁,往往后面一个请求的参数是前面一个请求的结果,于是经常需要在前面一个请求的响应中去发送第二个请求,从而造成“请求嵌套”的问题。如果层次比较多,代码可读性和效率都是问题。嵌套请求:利用RxJava的flatMap
操作符。
//第一个网络请求获取到tokenObservablelogin = EasyHttp.post(ComParamContact.Login.PATH) .params(ComParamContact.Login.ACCOUNT, "186****4275") .params(ComParamContact.Login.PASSWORD, MD5.encrypt4login("123456", AppConstant.APP_SECRET)) .sign(true) .timeStamp(true).execute(AuthModel.class); login.flatMap(new Function >() { @Override public ObservableSource apply(@NonNull AuthModel authModel) throws Exception { //获取到的token,给到第二个网络当参数。第二个网络开始请求 return EasyHttp.get("/v1/app/chairdressing/skinAnalyzePower/skinTestResult") .params("accessToken", authModel.getAccessToken())//这个地方只是举例,并不一定是需要accessToken .timeStamp(true) .execute(SkinTestResult.class); } }).subscribe(new ProgressSubscriber (this, mProgressDialog) { @Override public void onError(ApiException e) { super.onError(e); showToast(e.getMessage()); } @Override public void onNext(SkinTestResult skinTestResult) { showToast(skinTestResult.toString()); } });
本例中只是展示了2个接口的嵌套请求,flatMap其实是可以支持嵌套很多个接口请求
场景七:合并请求(zip)
zip合并请求就是指当一个页面有多个不同的数据来源,既就是有多个不同的网络请求接口,等待这些所有的接口都请求完成后才返回给订阅者,刷新界面等操作。
zip:使用一个函数组合多个Observable发射的数据集合,然后再发射这个结果.例如:一个页面有3个不同数据来源的网络请求接口,等待全部请求完成后才返回ObservablemobileObservable = EasyHttp.get("http://apis.juhe.cn/mobile/get") .params("phone", "18688994275") .params("dtype", "json") .params("key", "5682c1f44a7f486e40f9720d6c97ffe4") .execute(new CallClazzProxy , ResultBean>(ResultBean.class) { }); Observable searchObservable = EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); Observable
> listObservable = EasyHttp.get("http://news-at.zhihu.com/api/3/sections") .execute(new CallClazzProxy
>, List >(new TypeToken
>() { }.getType()) { }); //new Function3最后一个参数这里用的是List
1.zip的参数中前几个参数都表示数据源ObservableSource
,最后一个参数new Function会输出结果,前面如果有2个数据源就用Function2,3个数据源就用Function3. 2.Function的作用主要是会返回所有参与zip的结果内容,例如本例是有3个数据源(3个网络接口请求对应的Observable),返回的内容是new Function3<ResultBean, Content, List<SectionItem>, List<Object>>
其中ResultBean, Content, List<SectionItem>
是三个网络接口各自返回的内容。但是最后一个参数List<Object>
表示什么呢?这一个就表示最终要给订阅着subscribe
的内容,例如本例new BaseSubscriber<List<Object>>()
。这里我是将返回的结果用一个List集合存起来,然后返回给订阅者。最终想返回什么是根据自己的业务逻辑需求而定,并不一定就是List<Object>
,切记!切记!切记!
场景八:合并请求(merge)
刚才讲解了zip合并请求,这是合并请求的另一种场景实现方式merge,一定要注意merge和zip的区别,虽然都是合并多个请求,但是是有区别的,请注意使用场景. 本例中利用Rxjava的merge
、mergeDelayError
操作符。
mergeDelayError
操作符类似于merge
操作符,唯一不同就是如果在合并途中出现错误,不会立即发射错误通知,而是保留错误直到合并后的Observable将所有的数据发射完成,此时才会将onError提交给订阅者。例如:一个界面有两个网络接口请求 //这个请求故意延时5秒再发送->最后测试结果发现,并不是searchObservable等待mobileObservable5秒后再发送,目的验证无序的概念ObservablemobileObservable = Observable.timer(5, TimeUnit.SECONDS).flatMap(new Function >() { @Override public ObservableSource apply(@NonNull Long aLong) throws Exception { return EasyHttp.get("http://apis.juhe.cn/mobile/get") .params("phone", "18688994275") .params("dtype", "json") .params("key", "5682c1f44a7f486e40f9720d6c97ffe4") .execute(new CallClazzProxy , ResultBean>(ResultBean.class) { }); } }); Observable searchObservable = EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", "hello world") //采用代理 .execute(new CallClazzProxy , Content>(Content.class) { }); //Observable.merge(mobileObservable,searchObservable).subscribe(new BaseSubscriber () { Observable.mergeDelayError(mobileObservable,searchObservable).subscribe(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull Object object) { //为什么用Object接收,因为两个接口请求返回的数据类型不是一样的,如果是一样的就用具体的对象接收就可以了, // 不再需要instanceof麻烦的判断 if (object instanceof ResultBean) {//mobileObservable 返回的结果 //处理 ResultBean逻辑 } else if (object instanceof Content) { //处理 Content逻辑 } showToast(object.toString()); } });
其中提到一个概念:合并的Observable发射的数据交错,也就是发射的数据无序,怎么理解它呢?例如:代码中merge(mobileObservable,searchObservable)
合并的mobileObservable和searchObservable,其中mobileObservable在前,并不代表订阅的subscribe->onNext(@NonNull Object object)
就先返回来数据,也可能是合并的searchObservable数据先返回回来。
延伸讲解,刚才讲解了merge
操作符是无序的,如果想保证有序执行怎么办呢,采用Rxjava的concat
操作符
看完场景七和场景八:merge和zip的区别到底是什么呢?
merge和zip都是将多个Observalbe发射的数据项,合并到一个Observable中再发射出去。只是在发射的结果上有所不同。例如有3个网络请求的Observalbe,zip是等待这3个请求都完成后才一起返回,既onNext调用1次。merge是3个Observalbe分别返回,而且无序,既onNext调用3次,相当于把3个本来分散的网络请求。写在同一个地方合并起来执行。
merge、mergeDelayError都是合并,但是需要注意二者区别。
- merge合并的请求,如果有一个接口报错了,就立马报错,会终止整个流,另外的接口也不会请求。
- mergeDelayError合并的请求,如果有一个接口报错了,会延迟错误处理,后面的接口会继续执行没有被中断。
场景九:避免重复请求
这里主要是讲解RxEasyHttp如何结合rxbinding库compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
来使用。对rxbinding库不了解的,可以自己去学习一下,这里不详细介绍。在页面请求中可能操作太快导致同一个网络请求重复执行,这里结合view来说明,点击一个按钮去请求网络,点击太快就会重复执行-界面防抖。
throttleFirst
操作符。throttleFirst:会定期发射这个时间段里源Observable发射的第一个数据。例如:点击按钮去请求网络,1s内避免点击过快重复请求 RxView.clicks(view).throttleFirst(1, TimeUnit.SECONDS).flatMap(new Function>() { @Override public ObservableSource apply(@NonNull Object o) throws Exception { return EasyHttp.get("http://apis.juhe.cn/mobile/get") .params("phone", "18688994275") .params("dtype", "json") .params("key", "5682c1f44a7f486e40f9720d6c97ffe4") .execute(new CallClazzProxy , ResultBean>(ResultBean.class) { }); } }).subscribe(new BaseSubscriber () { @Override public void onError(ApiException e) { showToast(e.getMessage()); } @Override public void onNext(@NonNull ResultBean resultBean) { showToast(resultBean.toString()); } });
场景十:减少频繁的网络请求
这里也是主要讲解RxEasyHttp如何结合rxbinding库compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
来使用。对rxbinding库不了解的,可以自己去学习一下,这里不介绍用法。像即时搜索功能,在输入框中输入内容,实时搜索结果展示。
debounce
操作符。debounce:对源Observable每产生一个结果后,如果在规定的间隔时间内没有别的结果产生,则把这个结果提交给订阅者处理,否则忽略该结果。简单的理解就是:当N个结点发生的时间太靠近(即发生的时间差小于设定的值T),debounce就会自动过滤掉前N-1个结点。 例如:即时搜索,避免每输入(删除)一个字就做一次请求,500毫秒才让它去请求一次网络,这样可以避免数据混乱,也优了app性能。 Disposable mDisposable = RxTextView.textChangeEvents(mEditText) .debounce(500, TimeUnit.MILLISECONDS).filter(new Predicate() { @Override public boolean test(@NonNull TextViewTextChangeEvent textViewTextChangeEvent) throws Exception { // 过滤,把输入字符串长度为0时过滤掉 String key = textViewTextChangeEvent.text().toString(); //这里可以对key进行过滤的判断逻辑 return key.trim().length() > 0; } }).flatMap(new Function >() { @Override public ObservableSource apply(@NonNull TextViewTextChangeEvent textViewTextChangeEvent) throws Exception { String key = textViewTextChangeEvent.text().toString(); Log.d("test", String.format("Searching for: %s", textViewTextChangeEvent.text().toString())); return EasyHttp.get("/ajax.php") .baseUrl("http://fy.iciba.com") .params("a", "fy") .params("f", "auto") .params("t", "auto") .params("w", key) //采用代理 .execute(new CallClazzProxy , Content2>(Content2.class) { }); } }).subscribeWith(new BaseSubscriber () { @Override protected void onStart() { } @Override public void onError(ApiException e) { mTextView.setText(e.getMessage()); } @Override public void onNext(@NonNull Content2 content) { mTextView.setText(content.toString()); } }); //取消请求 //EasyHttp.cancelSubscription(mDisposable);
场景十一:网络数据缓存
网络数据缓存在RxEasyHttp网络库中已经封装了,也就是开发者在使用网络库过程中不必关心缓存具体的实现,通过讲解调用一些配置参数就可以实现需要的缓存功能了。这里主要讲解的是不使用okhttp和Retrofit的缓存,而是介绍自己如何封装缓存。RxEasyHttp网络库除了支持无缓存和默认的http缓存之外,又提供了其它6种场景缓存:
FIRSTREMOTE:先请求网络,请求网络失败后再加载缓存FIRSTCACHE:先加载缓存,缓存没有再去请求网络ONLYREMOTE:仅加载网络,但数据依然会被缓存ONLYCACHE:只读取缓存,缓存没有会返回nullCACHEANDREMOTE:先使用缓存,不管是否存在,仍然请求网络,CallBack会回调两次.CACHEANDREMOTEDISTINCT:先使用缓存,不管是否存在,仍然请求网络,CallBack回调不一定是两次,如果发现请求的网络数据和缓存数据是一样的,就不会再返回网络的回调,既回调一次。否则不相同仍然会回调两次。(目的是为了防止数据没有发生变化,也需要回调两次导致界面无用的重复刷新)如果对这部分源码敢兴趣的请查看源码,由于此部分涉及的知识点和内容比较多,篇幅问题,准备在下一篇文章中单独介绍网络数据缓存的各大场景实现,敬请期待!结语
到这里几大常用的RxEasyHttp网络库结合Rxjava2场景的例子已经介绍完了,也证明了RxEasyHttp网络库与Rxjava2是可以完美结合的,在实际开发运用中的场景肯定有很多种,例子是怎么也举不完的,需要灵活运用,如何写这一个自己的场景呢,我觉得需要如下几点:
- 充分了解自己的需求(需要的场景),明确目的(例如:我是想做一个网络轮询的,最终需要订阅者给我返回什么);
- 熟悉Rxjava操作符,看看哪个操作符是符合需求场景的,或者多个操作符串联起来是符合需求场景,了解每个操作符单个用法和订阅返回的值;
- 准备好网络请求,保证网络请求对象返回的是Observable,这样与Rxjava结合才能成为可能;
- 将Rxjava单个操作符或者多个操作符,与网络库的Observable进行结合,这里面可能是需要
flatMap
,Function
等操作;
文章难免有疏漏之处,如果您有任何疑议,请反馈给我,谢谢!同时大家有与网络相关更好更常用的场景请推荐给我,我会继续完善。喜欢请点赞,谢谢!
源码下载
对场景应用感兴趣的小伙伴们可以将Demo下载下来进行参考。
项目地址引用文字
欢迎加入我的圈子