阅读 4864

RxHttp 2000+star,协程请求,仅需三步

1、前言

Hello,各位小伙伴,3个月不见,我又回来了,这3个月里,一直致力于RxHttp项目的维护工作,截止本文发表,RxHttp已更新至v2.3.3版本,github上已突破2200+star,对于一个开源仅一年多一点的项目,可以说取得了不错的成绩,在这里,感谢大家的支持与肯定,RxHttp一定会越来越好,再次感谢大家。

在这之前,RxHttp已经有了一系列好用的操作符,如timeout、retry、map等等,这次,RxHttp给大家带来了更多更好用的操作符,如 flowOn、asFlow、filter、distinct、sort等等一系列使用频率非常高的操作符,并且支持自定义,接来下会一一介绍。

本文篇幅过长,大家可点赞收藏,喜欢的话,期望你可以给个star支持一下,开源不易,感激不尽。

RxHttp&RxLife 交流群:378530627

gradle依赖

apply plugin: 'kotlin-kapt'

android {
    kapt {
        arguments {
            //必须,传入okhttp版本号,可传入 3.12.0 ~ 4.7.2 任一版本号 (4.3.0除外)
            arg("rxhttp_okhttp", "4.7.2")
            //非必须,指定RxHttp类的包名
            arg("rxhttp_package", "rxhttp")
        }
    }
    //必须,java 8或更高
    compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }
}
dependencies {
    //以下3个为必须,
    implementation 'com.ljx.rxhttp:rxhttp:2.3.3'
    implementation 'com.squareup.okhttp3:okhttp:4.7.2' //rxhttp v2.2.2版本起,需要手动依赖okhttp
    kapt 'com.ljx.rxhttp:rxhttp-compiler:2.3.3' //生成RxHttp类,非kotlin项目,请使用annotationProcessor代替kapt
    
    //非必须,管理协程生命周期,页面销毁,关闭请求
    implementation 'com.ljx.rxlife:rxlife-coroutine:2.0.0' 

    //非必须,根据自己需求选择 RxHttp默认内置了GsonConverter
    implementation 'com.ljx.rxhttp:converter-jackson:2.3.3'
    implementation 'com.ljx.rxhttp:converter-fastjson:2.3.3'
    implementation 'com.ljx.rxhttp:converter-protobuf:2.3.3'
    implementation 'com.ljx.rxhttp:converter-simplexml:2.3.3'
}
复制代码

注:依赖完,记得rebuild(必须),才会生成RxHttp类

本文只介绍RxHttp与协程相关的部分,如果想了解RxJava相关,请阅读RxHttp 让你眼前一亮的Http请求框架

2、RxHttp 协程使用

2.1、请求三部曲

任意请求,任意返回值,皆遵循请求三部曲,掌握请求三部曲,就掌握了RxHttp的精髓,如下:

注:RxHttp2.2.0版本起,已完全剔除了RxJava,采用外挂的方法替代,支持RxJava2、RxJava3,详情查看RxHttp上手

协程请求三部曲代码表示

//协程,获取接口返回数据,以String类型返回
val str = RxHttp.get("/service/...")  //1、确定请求方式,可选get、postXxx等方法
    .toStr()                          //2、使用toXxx系列方法确定返回类型
    .await()                          //3、使用await方法拿到返回值
复制代码

协程请求三部曲详解

  • 第一步,选择get、postForm、postJson等方法来确定请求方式,随后便可通过add、addFile、addHeader/操作符来添加参数、文件、请求头,当然,我们还可以调用更多的操作符来满足业务需求,如setCacheXxx、upload等来设置缓存策略及监听上传进度等等

  • 第二步,调用toXxx系列方法来确定返回类型,常用的有toStr、toClass、toList,随后便可调用asFlow、retry、timeout、flowOn、filter、distinct、sort等30余个操作符来执行不同的业务逻辑,本文后续会一一介绍

  • 第三步,最后,只需调用await、tryAwait、awaitResult这三个中的任一操作符获取返回值即可,这一步,需要在协程环境中才能调用

如:我们要发送一个post表单请求,参数/文件/请求头各添加一个,并将返回数据反序列化为Student对象,其中超时时间设置为5秒;失败重试2次,每次间隔1秒,代码如下:

val student = RxHttp
    .postForm("/service/...")   //1、确定请求方式,可以选择get、postXxx等方法
    .add("key", "value")
    .addFile("file", File(".../1.png"))
    .addHeader("headerKey", "headerValue")
    .toClass<Student>()         //2、使用toXxx系列方法确定返回类型
    .timeout(5000)
    .retry(2, 1000)
    .await()                    //3、使用await方法拿到返回值
复制代码

2.2、BaseUrl处理

RxHttp通过@DefaultDomain、@Domain注解来配置默认域名及非默认域名,如下:

public class Url {

    @DefaultDomain //通过该注解设置默认域名
    public static String BASE_URL = "https://www.wanandroid.com";
    
    // name 参数在这会生成 setDomainToGoogleIfAbsent方法,可随意指定名称
    // className 参数在这会生成RxGoogleHttp类,可随意指定名称
    @Domain(name = "Google", className = "Google")
    public static String GOOGLE = "https://www.google.com";
}
复制代码

以上配置www.wanandroid.com为默认域名,www.google.com为非默认域名

多BaseUrl处理

//使用默认域名发请求
RxHttp.get("/service/...")
    .toSrt().await()
   
//使用google域名方式一:传入的url直接带上google域名
RxHttp.get("https://wwww.google.com/service/...")
    .toSrt().await()
    
//使用google域名方式二:调用setDomainToGoogleIfAbsent方法
RxHttp.get("/service/...")
    .setDomainToGoogleIfAbsent()
    .toSrt().await()
 
//使用google域名方式三:直接使用RxGoogleHttp类发送请求
RxGoogleHttp.get("/service/...")
    .toSrt().await()
复制代码

注意:手动传入的域名优先级别最高,其次是调用setDomainToXxx方法,最后才会使用默认域名

动态域名处理

//直接对url重新赋值即可,改完立即生效
Url.BASE_URL = "https://www.baidu.com";
RxHttp.get("/service/...")
    .toSrt().await()
//此时请求的url为 https://www.baidu.com/service/...
复制代码

2.3、业务code统一判断

我想大部分人的接口返回格式都是这样的

class Response<T> {
    var code = 0
    var msg : String? = null
    var data : T 
}
复制代码

拿到该对象的第一步就是对code做判断,如果code != 200(假设200代表数据正确),就会拿到msg字段给用户一些错误提示,如果等于200,就拿到data字段去更新UI,常规的写法是这样的

val response = RxHttp.get("/service/...")
    .toClass<Response<Student>>()
    .await()
if (response.code == 200) {
    //拿到data字段(Student)刷新UI
} else {
    //拿到msg字段给出错误提示
} 
复制代码

试想一下,一个项目少说也有十几个这样的接口,如果每个接口读取这么判断,就显得不够优雅,也可以说是灾难,相信也没有人会这么干。而且对于UI来说,只需要data字段即可,错误提示啥的我管不着。

那有没有什么办法,能直接拿到data字段,并且对code做出统一判断呢?有的,直接上代码

val student = RxHttp.get("/service/...")
    .toResponse<Student>() //调用此方法,直接拿到data字段,也就是Student对象   
    .await()  
//直接开始更新UI
复制代码

可以看到,这里调用了toResponse()方法,就直接拿到了data字段,也就是Student对象。

此时,相信很多人会有疑问,

  • 业务code哪里判断的?

  • 业务code非200时,msg字段怎么拿到?

为此,先来回答第一个问题,业务code哪里判断的?

其实toResponse()方法并不是RxHttp内部提供的,而是用户通过自定义解析器,并用@Parser注解标注,最后由注解处理器rxhttp-compiler自动生成的,听不懂?没关系,直接看代码

@Parser(name = "Response")
open class ResponseParser<T> : AbstractParser<T> {
    
    //以下两个构造方法是必须的
    protected constructor() : super()
    constructor(type: Type) : super(type)

    @Throws(IOException::class)
    override fun onParse(response: okhttp3.Response): T {
        val type: Type = ParameterizedTypeImpl[Response::class.java, mType] //获取泛型类型
        val data: Response<T> = convert(response, type)   //获取Response对象
        val t = data.data                             //获取data字段
        if (data.code != 200 || t == null) { //code不等于200,说明数据不正确,抛出异常
            throw ParseException(data.code.toString(), data.msg, response)
        }
        return t
    }
}
复制代码

上面代码只需要关注两点即可,

第一点,我们在类开头使用了@Parser注解,并为解析器取名为Response,所以就有了toResponse()方法(命名方式为:to + Parser注解里设置的名字);

第二点,我们在if语句里,对code做了判断,非200或者data为空时,就抛出异常,并带上了code及msg字段,所以我们在异常回调的地方就能拿到这两个字段

接着回答第二个问题,code非200时,如何拿到msg字段?直接上代码,看一个使用协程发送请求的完整案例

//当前环境在Fragment中
fun getStudent() {   
    //rxLifeScope在rxLife-coroutine库中,需要单独依赖                                 
    rxLifeScope.launch({    //通过launch方法开启一个协程                           
        val student = RxHttp.get("/service/...") 
            .toResponse<Student>()
            .await()                 
    }, {                                               
        //异常回调,这里的it为Throwable类型                                        
        val code = it.code                            
        val msg = it.msg                              
    })                                                
}                                                     
复制代码

注:RxLifeScope 是 RxLife-Coroutine库中的类,本文后续会详细介绍

上面的代码,在异常回调中便可拿到code及msg字段,需要注意的是,it.codeit.msg是我为Throwable类扩展的两个属性,代码如下:

val Throwable.code: Int
    get() {
        val errorCode = when (this) {
            is HttpStatusCodeException -> this.statusCode //Http状态码异常
            is ParseException -> this.errorCode     //业务code异常
            else -> "-1"
        }
        return try {
            errorCode.toInt()
        } catch (e: Exception) {
            -1
        }
    }

val Throwable.msg: String
    get() {
        return if (this is UnknownHostException) { //网络异常
            "当前无网络,请检查你的网络设置"
        } else if (
            this is SocketTimeoutException  //okhttp全局设置超时
            || this is TimeoutException     //rxjava中的timeout方法超时
            || this is TimeoutCancellationException  //协程超时
        ) {
            "连接超时,请稍后再试"
        } else if (this is ConnectException) {
            "网络不给力,请稍候重试!"
        } else if (this is HttpStatusCodeException) {               //请求失败异常
            "Http状态码异常"
        } else if (this is JsonSyntaxException) {  //请求成功,但Json语法异常,导致解析失败
            "数据解析失败,请检查数据是否正确"
        } else if (this is ParseException) {       // ParseException异常表明请求成功,但是数据不正确
            this.message ?: errorCode   //msg为空,显示code
        } else {
            "请求失败,请稍后再试"
        }
    }
复制代码

到这,业务code统一判断就介绍完毕,上面的代码,大部分人都可以简单修改后,直接用到自己的项目上,如ResponseParser解析器,只需要改下if语句的判断条件即可

2.4、操作符介绍

map 转换符号

map操作符很好理解,RxJava及协程的Flow都有该操作符,功能都是一样,用于转换对象,如下:

val student = RxHttp.postForm("/service/...")
    .toStr()
    .map { it.length }  //String转Int                        
    .tryAwait()     //这里返回 Student? 对象,即有可能为空  
复制代码

timeout 超时

OkHttp提供了全局的读、写及连接超时,有时我们也需要为某个请求设置不同的超时时长,此时就可以用到RxHttp的timeout(Long)方法,如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(3000)      //超时时长为3s                           
    .await()                       
复制代码

retry 失败重试

OkHttp为我们提供了全局的失败重试机制,然而,这远远不能满足我们的需求,比如,我就部分接口需要失败重试,而不是全局的;我需要根据某些条件来判断是否需要重试;亦或者我需要周期性重试,即间隔几秒后重试等等

如我们需要在网络出现异常时,重试2次,每次间隔1秒,代码如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .retry(2, 1000) {            //重试2次,每次间隔1s                       
        it is ConnectException   //如果是网络异常就重试     
    }                                             
    .await()                     
复制代码

retry()方法共有3个参数,分别是重试次数、重试周期、重试条件,都有默认值,3个参数可以随意搭配,如下:

/**
 * 失败重试,该方法仅在使用协程时才有效
 * @param times  重试次数, 默认Int.MAX_VALUE 代表不断重试
 * @param period 重试周期, 默认为0, 单位: milliseconds
 * @param test   重试条件, 默认为true,只要出现异常就重试
 */
fun retry(
    times: Int = Int.MAX_VALUE,
    period: Long = 0,
    test: suspend (Throwable) -> Boolean = { true }
)
复制代码

filter 过滤操作

如果服务器返回列表数据,则我们可对列表进行过滤操作,如下:

val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .filter{ it.age > 20 }   //过滤年龄大于20岁的学生
    .await()                     
复制代码

还可以选用filterTo操作符,将过滤后的数据添加到指定列表中,如下:

val list = mutableListOf<Student>()
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .filterTo(list){ it.age > 20 }   //过滤年龄大于20岁的学生
    .await()  //此时返回的列表对象就是我们传入的列表对象
    
复制代码

distinct 去重

该操作符可以对服务器返回的列表,做去重操作,如下:

//根据Student对象的hashCode去重
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .distinct()   
    .await()    

//根据Student对象的id去重
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .distinctBy { it.id }
    .await()   
    
//将去重后的数据添加到指定列表中,并且去重时,会对指定列表数据做判断
val list = mutableListOf<Student>()
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .distinctTo(list) { it.id }
    .await()   
复制代码

sort 排序

排序有sortXxx、sortedXxx两大类型操作符,区别在于sortXxx在列表内排序,排序完,返回自身,而sortedXxx在列表外排序,排序完,返回新的列表,这里只对sortXxx介绍,如下:

//根据id顺序排序
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .sortBy { it.id }   
    .await() 
    
//根据id、age两个字段顺序排序,id优先,其次age
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .sortBy({ it.id }, { it.age })  
    .await() 

//返回两个排序对象,自行实现排序规则
val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .sortWith { student1, student2 ->             
        student1.id.compareTo(student2.id)        
    }                                             
    .await() 
复制代码

flowOn 指定上游所在线程

该操作符跟Flow里面的flowOn操作符一样,用于指定上游所在线程,如下:

val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .sortBy { it.id }        //IO线程执行
    .flowOn(Dispatchers.IO)
    .distinctBy { it.id }    //Default线程执行
    .flowOn(Dispatchers.Default)
    .filter{ it.age > 20 }   //IO线程执行
    .flowOn(Dispatchers.IO)
    .flowOn(Dispatchers.Default)
    .await() 
复制代码

asFlow 转Flow对象

如果你喜欢kotlin的flow流,那么asFlow 就派上用场了,如下:

RxHttp.postForm("/service/...")
    .toList<Student>()
    .asFlow()
    .collect {       
        //这里拿到List<Student>对象         
    }                
复制代码

注意:使用asFlow操作符后,需要使用collect替代await操作符

subList、take 截取列表

subList用于截取某段列表,截取范围越界,则抛出越界异常;take用于从0开始,取n个数据,不足n个时,返回全部,如下:

val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .subList(110)  //截取9个数据
    .take(5)         //从9个中取前5个
    .await()               
复制代码

async 异步操作

如果我们由两个请求需要并行时,就可以使用该操作符,如下:

//同时获取两个学生信息
suspend void initData() {
  val asyncStudent1 = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .async(this)   //this为CoroutineScope对象,这里会返回Deferred<Student>  
    
  val asyncStudent2 = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .async(this)   //this为CoroutineScope对象,这里会返回Deferred<Student>  

  //随后调用await方法获取对象    
  val student1 = asyncStudent1.await()
  val student2 = asyncStudent2.await()
} 
复制代码

delay、startDelay 延迟

delay操作符是请求结束后,延迟一段时间返回;而startDelay操作符则是延迟一段时间后再发送请求,如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .delay(1000)      //请求回来后,延迟1s返回                         
    .await()       
    
val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .startDelay(1000)     //延迟1s后再发送请求       
    .await()     
复制代码

onErrorReturn、onErrorReturnItem异常默认值

有些情况,我们不希望请求出现异常时,直接走异常回调,此时我们就可以通过两个操作符,给出默认的值,如下:

//根据异常给出默认值
val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(100)      //超时时长为100毫秒  
    .onErrorReturn {
        //如果时超时异常,就给出默认值,否则,抛出原异常
        return@onErrorReturn if (it is TimeoutCancellationException)
            Student()                                              
        else                                                        
            throw it                                                
    }
    .await()
    
//只要出现异常,就返回默认值
val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(100)      //超时时长为100毫秒  
    .onErrorReturnItem(Student())
    .await()
复制代码

awaitResult 返回kotlin.Result

这个就直接看代码吧

val result: Result<Student> = RxHttp
    .postForm("/service/...")
    .toClass<Student>()
    .awaitResult()
if (result.isSuccess) { 
    //请求成功,拿到Student对象
    val student = result.getOrThrow()            
} else { 
    //请求出现异常,拿到Throwable对象
    val throwable = result.exceptionOrNull()     
}                                                
复制代码

拿到kotlin.Result对象后,我们需要判断请求成功与否,随后在执行相关逻辑

tryAwait 异常返回null

tryAwait会在异常出现时,返回null,如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(100)      //超时时长为100毫秒                        
    .tryAwait()     //这里返回 Student? 对象,如果出现异常,那它就是null  
复制代码

自定义操作符

RxHttp内置了一系列强大又好用的操作符,然而肯定满足不了所有的业务场景,此时我们就可以考虑自定义操作符

自定义takeLast操作符

如我们有这样一个需求,自定义需要在列表尾部取n条数据,不足n条,返回全部

前面我们介绍了take操作符,它是从0开始,取n条数据,如果不足n条,则全部返回,来看看源码

fun <T> IAwait<out Iterable<T>>.take(
    count: Int
): IAwait<List<T>> = newAwait {
    await().take(count)
}
复制代码

代码解读,

1、IAwait是一个接口,如下:

interface IAwait<T> {

    suspend fun await(): T
}
复制代码

该接口仅有一个await()方法,返回声明的T

2、newAwait操作符,只是创建了一个IAwait接口的实现而已,如下:

inline fun <T, R> IAwait<T>.newAwait(
    crossinline block: suspend IAwait<T>.() -> R
): IAwait<R> = object : IAwait<R> {

    override suspend fun await(): R {
        return this@newAwait.block()
    }
}
复制代码

3、由于我们是为IAwait<out Iterable<T>>对象扩展的take方法,故在内部,我们调用await()方法它返回Iterable<T>对象,最后就执行Iterable<T>对象的扩展方法take(Int)获取从0开是的n条数据,take(Int)是系统提供的方法,源码如下:

public fun <T> Iterable<T>.take(n: Int): List<T> {
    require(n >= 0) { "Requested element count $n is less than zero." }
    if (n == 0) return emptyList()
    if (this is Collection<T>) {
        if (n >= size) return toList()
        if (n == 1) return listOf(first())
    }
    var count = 0
    val list = ArrayList<T>(n)
    for (item in this) {
        list.add(item)
        if (++count == n)
            break
    }
    return list.optimizeReadOnlyList()
}
复制代码

ok,回到前面的话题,如何自定义一个操作,实现在列表尾部取n条数据,不足n条,返回全部

看了上面的take(int)源码,我们就可以很轻松的写出如下代码:

fun <T> IAwait<out List<T>>.takeLast(
    count: Int
): IAwait<List<T>> = newAwait {
    await().takeLast(count)
}
复制代码

首先,我们对IAwait<out List<T>>扩展了takeLast(Int)方法,随后调用newAwait创建了IAwait接口的实例对象,接着调用await()方法返回List<T>对象,最后调用系统为List<T>扩展的takeLast(Int)方法

定义好了,我们就可直接使用,如下:

val students = RxHttp.postForm("/service/...")
    .toList<Student>()
    .takeLast(5)         //取列表尾部5条数据,不足时,全部返回
    .await()               
复制代码

以上操作符随意搭配

以上操作符,可随意搭配使用,但调用顺序的不同,产生的效果也不一样,这里先告诉大家,以上操作符仅会对上游代码产生影响。

timeout及retry

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(50)
    .retry(2, 1000) { it is TimeoutCancellationException }                                  
    .await()                       
复制代码

以上代码,只要出现超时,就会重试,并且最多重试两次。

但如果timeoutretry互换下位置,就不一样了,如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .retry(2, 1000) { it is TimeoutCancellationException }       
    .timeout(50)                                  
    .await()                       
复制代码

此时,如果50毫秒内请求没有完成,就会触发超时异常,并且直接走异常回调,不会重试。为什么会这样?原因很简单,timeout及retry操作符,仅对上游代码生效。如retry操作符,下游的异常是捕获不到的,这就是为什么timeout在retry下,超时时,重试机制没有触发的原因。

在看timeoutstartDelay操作符

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .startDelay(2000)       
    .timeout(1000)                                  
    .await()                       
复制代码

以上代码,必定会触发超时异常,因为startDelay,延迟了2000毫秒,而超时时长只有1000毫秒,所以必定触发超时。 但互换下位置,又不一样了,如下:

val student = RxHttp.postForm("/service/...")
    .toResponse<Student>()
    .timeout(1000)    
    .startDelay(2000)       
    .await()                       
复制代码

以上代码正常情况下,都能正确拿到返回值,为什么?原因很简单,上面说过,操作符只会对上游产生影响,下游的startDelay延迟,它是不管的,也管不到。

3、上传/下载

RxHttp对文件的优雅操作是与生俱来的,在协程的环境下,依然如此,没有什么比代码更具有说服力,直接上代码

3.1、文件上传

 val result = RxHttp.postForm("/service/...")  
     .addFile("file", File("xxx/1.png"))        //添加单个文件   
     .addFile("fileList", ArrayList<File>())    //添加多个文件
     .toResponse<String>()
     .await()                   
复制代码

只需要通过addFile系列方法添加File对象即可,就是这么简单粗暴,想监听上传进度?简单,再加一个upload操作符即可,如下:

val result = RxHttp.postForm("/service/...")                               
    .addFile("file", File("xxx/1.png"))                                     
    .addFile("fileList", ArrayList<File>())                                 
    .upload(this) {    //此this为CoroutineScope对象,即当前协程对象  
        //it为Progress对象                                                
        val process = it.progress         //已上传进度  0-100                       
        val currentSize = it.currentSize  //已上传size,单位:byte                 
        val totalSize = it.totalSize      //要上传的总size  单位:byte           
    }                                                               
    .toResponse<String>()        
    .await()                                                                                                                       
复制代码

我们来看下upload方法的完整签名,如下:

/**
 * 调用此方法监听上传进度                                                    
 * @param coroutine  CoroutineScope对象,用于开启协程,回调进度,进度回调所在线程取决于协程所在线程
 * @param progress 进度回调  
 * 注意:此方法仅在协程环境下才生效                                         
 */
fun RxHttpFormParam.upload(
    coroutine: CoroutineScope? = null, 
    progress: (Progress) -> Unit
):RxHttpFormParam
复制代码

3.2、文件下载

接着再来看看下载,直接贴代码

val localPath = "sdcard//android/data/..../1.apk" 
val student = RxHttp.get("/service/...")     
    .toDownload(localPath)  //下载需要传入本地文件路径   
    .await()
复制代码

下载调用toDownload(String)方法,传入本地文件路径即可,要监听下载进度?也简单,如下:

val localPath = "sdcard//android/data/..../1.apk"  
val student = RxHttp.get("/service/...")      
    .toDownload(localPath, this) {   //此this为CoroutineScope对象     
        //it为Progress对象
        val process = it.progress        //已下载进度 0-100
        val currentSize = it.currentSize //已下载size,单位:byte
        val totalSize = it.totalSize     //要下载的总size 单位:byte                                           
    }     
    .await()
复制代码

看下toDownload方法完整签名

/**
 * @param destPath 本地存储路径
 * @param coroutine CoroutineScope对象,用于开启协程,回调进度,进度回调所在线程取决于协程所在线程
 * @param progress 进度回调
 */
fun IRxHttp.toDownload(
    destPath: String,
    coroutine: CoroutineScope? = null,
    progress: (Progress) -> Unit
): IAwait<String>
复制代码

如果你需要断点下载,也是可以的,一行代码的事,如下:

val localPath = "sdcard//android/data/..../1.apk"                        
val student = RxHttp.get("/service/...")                            
    .setRangeHeader(1000, 300000)   //断点下载,设置下载起始/结束位置                                     
    .toDownload(localPath, this) { //此this为CoroutineScope对象           
        //it为Progress对象                                                  
        val process = it.progress        //已下载进度 0-100                          
        val currentSize = it.currentSize //已下size,单位:byte               
        val totalSize = it.totalSize     //要下的总size 单位:byte                 
    }      
    .await()
复制代码

老规则,看下setRangeHeader完整签名

/**                                                                                      
 * 设置断点下载开始/结束位置                                                                         
 * @param startIndex 断点下载开始位置                                                            
 * @param endIndex 断点下载结束位置,默认为-1,即默认结束位置为文件末尾                                           
 * @param connectLastProgress 是否衔接上次的下载进度,该参数仅在带进度断点下载时生效                                
 */                                                                                      
fun setRangeHeader (
    startIndex: Long, 
    endIndex: Long = 0L, 
    connectLastProgress: Boolean = false
)
复制代码

到这,RxHttp协程的基础Api基本介绍完毕,那么问题了,以上介绍的Api都依赖与协程环境,那我这么开启协程呢?亦或者说,我对协程不是很懂,你只要保证安全的前提下,告诉怎么用就行了,ok,那下面如何安全的开启一个协程,做到自动异常捕获,且页面销毁时,自动关闭协程及请求

4、协程开启及关闭

此时就要引入本人开源的另一个库RxLife-Coroutine,用于开启/关闭协程,并自动异常捕获,依赖如下:

implementation 'com.ljx.rxlife:rxlife-coroutine:2.0.0'
复制代码

本文在介绍业务code统一处理的时候,我们用到rxLifeScope属性开启协程,那这个是什么类型呢?看代码

val ViewModel.rxLifeScope: RxLifeScope
    get() {
        val scope: RxLifeScope? = this.getTag(JOB_KEY)
        if (scope != null) {
            return scope
        }
        return setTagIfAbsent(JOB_KEY, RxLifeScope())
    }

val LifecycleOwner.rxLifeScope: RxLifeScope
    get() = lifecycle.rxLifeScope
复制代码

可以看到,我们为ViewModelLifecycleOwner都扩展了一个名为rxLifeScope的属性,类型为RxLifeScope,ViewModel相信大家都知道了,这里就简单讲一下LifecycleOwner接口,我们的Fragment及FragmentActivity都实现了LifecycleOwner接口,而我们的Activity一般继承于AppCompatActivity,而AppCompatActivity继承于FragmentActivity,所以我们在FragmentActivity/Fragment/ViewModel环境下,可以直接使用rxLifeScope开启协程,如下:

rxLifeScope.lanuch({
    //协程代码块,运行在UI线程
}, {
    //异常回调,协程代码块出现任何异常,都会直接走这里    
})
复制代码

通过这种方式开启的协程,会在页面销毁时,会自动关闭协程,当然,如果你的协程代码块里还有RxHttp请求的代码,协程关闭的同时,也是关闭请求,所以在这种情况下,只需要知道如何开启协程就行,其它一律不管。

现在,我们来看下rxLifeScope.lanuch方法的完整签名

/**                                                         
 * @param block     协程代码块,运行在UI线程                           
 * @param onError   异常回调,运行在UI线程                            
 * @param onStart   协程开始回调,运行在UI线程                          
 * @param onFinally 协程结束回调,不管成功/失败,都会回调,运行在UI线程             
 */                                                         
fun launch(                                                 
    block: suspend CoroutineScope.() -> Unit,               
    onError: ((Throwable) -> Unit)? = null,                 
    onStart: (() -> Unit)? = null,                          
    onFinally: (() -> Unit)? = null                         
): Job                                                    
复制代码

可以看到,不仅有失败回调,还有开始及结束回调,这对于我们发请求来说,真的非常方便,如下:

rxLifeScope.launch({                                      
    //协程代码块                                              
    val students = RxHttp.postJson("/service/...")
        .toResponse<List<Student>>()
        .await()   
    //可以直接更新UI                   
}, {                                                      
    //异常回调,这里可以拿到Throwable对象                             
}, {                                                     
    //开始回调,可以开启等待弹窗                                      
}, {                                                     
    //结束回调,可以销毁等待弹窗                                      
})                                                       
复制代码

以上代码均运行在UI线程中,请求回来后,便可直接更新UI

也许你还有疑问,我在非FragmentActivity/Fragment/ViewModel环境下,如何开启协程,又如何关闭,其实也很简单,如下:

val job = RxLifeScope().launch({                                        
    val students = RxHttp.postJson("/service/...")
        .toResponse<List<Student>>()
        .await()            
}, {                                                
    //异常回调,这里可以拿到Throwable对象                       
}, {                                               
    //开始回调,可以开启等待弹窗                                
}, {                                               
    //结束回调,可以销毁等待弹窗                                
})                                                 
job.cancel()  //关闭协程                                                                                        
复制代码

以上代码,需要注意两点,第一,我们需要手动创建RxLifeScope()对象,随后开启协程;第二,开启协程后,可以拿到Job对象,我们需要通过该对象手动关闭协程。其它就没啥区别了。

5、协程多任务处理

我们知道,协程最大的优势就是:能以看起来同步的代码,写出异步的逻辑,这使得我们可以非常优雅的实现多任务场景,比如多请求的并行/串行

5.1、协程串行多个请求

假设,我们有这么一种场景,首先获取Student对象,随后通过studentId获取学生的家庭成员列表,后者依赖于前者,这是典型的串行场景

看看通过协程如何解决这个问题,如下:

class MainActivity : AppCompatActivity() {
    //启动协程,发送请求
    fun sendRequest() {
        rxLifeScope.launch({
            //当前运行在协程中,且在主线程运行
            val student = getStudent()
            val personList = getFamilyPersons(student.id) //通过学生Id,查询家庭成员信息
            //拿到相关信息后,便可直接更新UI,如:
            tvName.text = student.name
        }, {
                //出现异常,就会到这里,这里的it为Throwable类型
            it.show("发送失败,请稍后再试!") //show方法是在Demo中扩展的方法
        })
    }

    //挂断方法,获取学生信息
    suspend fun getStudent(): Student {
        return RxHttp.get("/service/...")
            .add("key", "value")
            .addHeader("headKey", "headValue")
            .toClass<Student>()
            .await()
    }

    //挂断方法,获取家庭成员信息
    suspend fun getFamilyPersons(studentId: Int): List<Person> {
        return RxHttp.get("/service/...")
            .add("studentId", "studentId")
            .toClass<List<Person>>()
            .await()
    }
}
复制代码

我们重点看下协程代码块,首先通过第一个请求拿到Student对象,随后拿到studentId,发送第二个请求获取学习家庭成员列表,拿到后,便可以直接更新UI,怎么样,是不是看起来同步的代码,写出了异步的逻辑。

串行请求中,只要其中一个请求出现异常,协程便会关闭(同时也会关闭请求),停止执行剩下的代码,接着走异常回调

5.2、协程并行多个请求

请求并行,在现实开发中,也是家常便饭,在一个Activity中,我们往往需要拿到多种数据来展示给用户,而这些数据,都是不同接口下发的。

如我们有这样一个页面,顶部是横向滚动的Banner条,Banner条下面展示学习列表,此时就有两个接口,一个获取Banner条列表,一个获取学习列表,它们两个互不依赖,便可以并行执行,如下:

class MainActivity : AppCompatActivity() {
    //启动协程,发送请求
    fun sendRequest() {
        rxLifeScope.launch({
            //当前运行在协程中,且在主线程运行
            val asyncBanner = getBanners(this) //这里返回Deferred<List<Banner>>对象
            val asyncPersons = getStudents(this) //这里返回Deferred<List<Student>>对象
            val banners = asyncBanner.await()           //这里返回List<Banner>对象
            val students = asyncPersons.await()         //这里返回List<Student>对象
            //开始更新UI

        }, {
                //出现异常,就会到这里,这里的it为Throwable类型
            it.show("发送失败,请稍后再试!") //show方法是在Demo中扩展的方法
        })
    }

    //挂断方法,获取学生信息
    suspend fun getBanners(scope: CoroutineScope): Deferred<List<Banner>> {
        return RxHttp.get("/service/...")
            .add("key", "value")
            .addHeader("headKey", "headValue")
            .toClass<List<Banner>>()
            .async(scope)  //注意这里使用async异步操作符
    }

    //挂断方法,获取家庭成员信息
    suspend fun getStudents(scope: CoroutineScope): Deferred<List<Student>> {
        return RxHttp.get("/service/...")
            .add("key", "value")
            .toClass<List<Student>>()
            .async(scope) //注意这里使用async异步操作符
    }
}
复制代码

在上述代码的两个挂断方法中,均使用了async异步操作符,此时这两个请求就并行发送请求,随后拿到Deferred<T>对象,调用其await()方法,最终拿到Banner列表及Student列表,最后便可以直接更新UI。

划重点

并行跟串行一样,如果其中一个请求出现了异常,协程便会自动关闭(同时关闭请求),停止执行剩下的代码,接着走异常回调。如果想多个请求互不影响,就可以使用上面介绍的onErrorReturnonErrorReturnItem操作符,出现异常时,给出一个默认对象,又或者使用tryAwait操作符获取返回值,出现异常时,返回null,这样就不会影响其它请求的执行。

6、总结

看完本文,相信你已经领悟到了RxHttp优雅及简便,业务code的统一处理,失败重试、超时、文件上传/下载及进度监听,到后面rxLifeScope协程的开启/关闭/异常处理/多任务处理,一切都是那么的优雅。

其实,RxHttp远不止这些,本文只是讲解了RxHttp与协程相关的东西,更多优雅的功能,如:公共参数/请求头的添加、请求加解密、缓存等等,请查看

RxHttp 让你眼前一亮的Http请求框架

RxHttp 全网Http缓存最优解

最后,开源不易,写文章更不易,还需劳烦大家给本文点个赞,可以的话,再给个star,我将感激不尽,🙏🙏🙏🙏🙏🙏🙏🙏🙏🙏🙏🙏