Kotlin协程请求网络

本文记录一下Kotlin协程如何配合其他网络请求框架来进行网络请求。其中涉及的底层原理暂时不去关注。

本篇文章中使用到的接口来自wanandroid提供的公开接口。

相关代码可以参考KotlinAndroid中的CoroutineOkHttpNetActivity和CoroutineRetrofitNetActivity。

加入依赖

//使用kotlin的依赖
implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.3.71"

//在Android中使用协程需要添加此依赖
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.0'

Kotlin协程配合OkHttp

先初始化OkHttpClient

private lateinit var client: OkHttpClient
private lateinit var builder: OkHttpClient.Builder

override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_coroutine_net)

        builder = OkHttpClient.Builder()
                .readTimeout(5000, TimeUnit.MILLISECONDS)
                .writeTimeout(10000, TimeUnit.MILLISECONDS)

        client = builder
                .build()
}

OkHttp正常发起网络请求

private fun normalRequest() {
    //请求公众号列表
    val request = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()

    client.newCall(request).enqueue(object : okhttp3.Callback {

        override fun onFailure(call: okhttp3.Call, e: IOException) {
            Log.d(TAG, "onFailure: ${e.message}")
        }

        override fun onResponse(call: okhttp3.Call, response: okhttp3.Response) {
            val string = response.body()?.string()
            Log.d(TAG, "正常请求 onResponse: $string")
            //切换到主线程
            runOnUiThread {
                tvResult.text = "正常请求 onResponse: $string"
            }
        }
    })
}

协程配合OkHttp正常发起网络请求

1. 首先给okhttp3.Call添加一个扩展函数，就是对okhttp3.Call的enqueue方法做了一个包装。

suspend fun okhttp3.Call.awaitResponse(): okhttp3.Response {

    return suspendCancellableCoroutine {

      it.invokeOnCancellation {
            //当协程被取消的时候，取消网络请求
            cancel()
        }

        enqueue(object : okhttp3.Callback {
            override fun onFailure(call: okhttp3.Call, e: IOException) {
                it.resumeWithException(e)
            }

            override fun onResponse(call: okhttp3.Call, response: okhttp3.Response) {
                it.resume(response)
            }
        })
    }
}

然后我们在Activity中使用

class CoroutineOkHttpNetActivity : AppCompatActivity(), CoroutineScope by MainScope() {

    override fun onDestroy() {
        super.onDestroy()
        //取消所有的任务
        this.cancel()
    }

}

我们实现了CoroutineScope接口，然后我们将CoroutineScope的实现委托给MainScope。注意要在onDestroy的时候取消所有的任务。

发起请求

 //注释1处
val exceptionHandler = CoroutineExceptionHandler { coroutineContext, throwable ->
    Log.d(TAG, "coroutine: error ${throwable.message}")
}

private fun coroutineRequest() {
    //请求公众号列表
    val request1 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()

    //使用exceptionHandler
    launch(exceptionHandler) {
        //注释2处
        val response = client.newCall(request1).awaitResponse()
        //注释3处
        val string = getString(response)
        //合并两次请求的结果更新UI
        tvResult.text = "协程请求 onResponse: $string"
    }
}

注释1处，我们定义了一个CoroutineExceptionHandler，用来捕获处理异常信息。关于异常处理的内容后面会单独写一篇文章。

注释2处，调用awaitResponse方法发起网络请求。

注释3处，因为okhttp3.ResponseBody的string方法是一个IO操作，可能会比较耗时。所以我们通过一个挂起函数将这个操作放在后台线程执行。

private suspend fun getString(response: Response): String {
    return withContext(Dispatchers.IO) {
        response.body()?.string() ?: "empty string"
    }
}

我们在开发过程中，通常会遇到这样的场景：发起两次网络请求，第二次网络请求要依赖于第一次网络请求的结果。那我们该怎么写呢？

我们先获取公众号列表，然后查看列表中第一个公众号的历史数据。

private fun coroutineRequest() {
    val request1 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()

    launch(exceptionHandler) {
        //第一个网络请求
        val response1 = client.newCall(request1).awaitResponse()
        val string1 = getString(response1)
        val wxArticleResponse = JsonUtilKt.instance.toObject(string1, WxArticleResponse::class.java)

        //第二个网络请求依赖于第一个网络请求结果
        val firstWxId = wxArticleResponse?.data?.get(0)?.id ?: return@launch
        //第二个网络请求
        val request2 = Request.Builder()
                .url("https://wanandroid.com/wxarticle/list/${firstWxId}/1/json")
                .build()
        val response2 = client.newCall(request2).awaitResponse()
        val string2 = getString(response2)

        tvResult.text = "协程请求 onResponse: ${string2}"
    }
}

我们可以看到两次网络请求可以顺序书写下来，没有回调嵌套，非常简洁。

我们有时还会遇到这样的场景，需要将两次网络请求的结果合并到一起再进行下一步的操作，但是两个网络请求是没有依赖关系的，可以并发进行。我们该怎么写呢？

我们两次获取公众号列表，然后将结果合并起来。

如果按照上面的例子，我们可以这样写：

private fun coroutineRequest() {
    val request1 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()
    val request2 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()
    launch(exceptionHandler) {
        val startTime = System.currentTimeMillis()
        //注释1处，发起两次请求    
        val response1 = client.newCall(request1).awaitResponse()
        val response2 = client.newCall(request2).awaitResponse()

        Log.d(TAG, "coroutineRequest: 网络请求消耗时间：${System.currentTimeMillis() - startTime}")
        val string1 = getString(response1)
        val string2 = getString(response2)
            
        //合并两次请求的结果更新UI
        tvResult.text = "协程请求 onResponse: ${string1 + string2}"
    }
}

这种写法是不合适的，为什么呢？因为这两次请求是顺序执行的。在我的设备上多次执行coroutineRequest方法，两次网络请求最少的耗时时间是70毫秒。打印日志：

D/CoroutineOkHttpNetActiv: coroutineRequest: 网络请求消耗时间：70

正确的写法，并行发起请求。

private fun coroutineRequest3() {
    val request1 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()
    val request2 = Request.Builder()
            .url("https://wanandroid.com/wxarticle/chapters/json")
            .build()

    launch(exceptionHandler) {
        val startTime = System.currentTimeMillis()
        //两次网络请求没有依赖关系，可以并发请求
        val deferred1 = async { client.newCall(request1).awaitResponse() }
        val deferred2 = async { client.newCall(request2).awaitResponse() }
        val response1 = deferred1.await()
        val response2 = deferred2.await()
        Log.d(TAG, "coroutineRequest: 并发网络请求消耗时间：${System.currentTimeMillis() - startTime}")
        val string1 = getString(response1)
        val string2 = getString(response2)
        tvResult.text = "协程请求 onResponse: ${string1 + string2}"
    }
}

在我的设备上，并发执行两次网络请求最少的耗时时间是45毫秒。打印日志：

D/CoroutineOkHttpNetActiv: coroutineRequest: 并发网络请求消耗时间：45

Kotlin协程配合Retrofit

定义接口方法

interface ApiService {

    @GET("wxarticle/chapters/json")
    fun getWxarticle(): Call<WxArticleResponse>

}

构建接口实例

private val apiService = Retrofit.Builder()
            .baseUrl("https://www.wanandroid.com/")
            .addConverterFactory(GsonConverterFactory.create())
            .build().create(ApiService::class.java)

Retrofit正常请求

private fun normalRequest() {
    apiService.getWxarticle().enqueue(object : retrofit2.Callback<WxArticleResponse> {
        override fun onFailure(call: Call<WxArticleResponse>, t: Throwable) {
            Log.d(TAG, "onFailure: ${t.message}")
        }

        override fun onResponse(call: Call<WxArticleResponse>, response: retrofit2.Response<WxArticleResponse>) {
            if (response.isSuccessful) {
                val wxArticleResponse = response.body()
                val sb = StringBuilder("Retrofit正常请求：\n")
                wxArticleResponse?.data?.forEach {
                    sb.append(it.name)
                    sb.append("\n")
                }
                tvResult.text = sb.toString()
            }
        }
    })
}

协程配合Retrofit（2.6.0以下版本）正常网络请求

1. 首先给Retrofit.Call类添加一个扩展函数，就是对retrofit2.Call的enqueue方法做了一个包装。

suspend fun <T : Any?> Call<T>.awaitResponse(): T {

    return suspendCancellableCoroutine {
        enqueue(object : Callback<T> {
            override fun onFailure(call: Call<T>, t: Throwable) {
                it.resumeWithException(t)
            }

            override fun onResponse(call: Call<T>, response: Response<T>) {
                if (response.isSuccessful) {
                    val body = response.body()
                    if (body != null) {
                        it.resume(body)
                    } else {
                        it.resumeWithException(Throwable(response.toString()))
                    }

                } else {
                    it.resumeWithException(Throwable(response.toString()))
                }
            }
        })
    }
}

在Activity中使用

class CoroutineOkHttpNetActivity : AppCompatActivity(), CoroutineScope by MainScope() {

    override fun onDestroy() {
        super.onDestroy()
        //取消所有的任务
        this.cancel()
    }

}

我们实现了CoroutineScope接口，然后我们将CoroutineScope的实现委托给MainScope。注意要在onDestroy的时候取消所有的任务。

发起请求

private fun coroutineRequest() {
    launch(exceptionHandler) {
         //需要借助Retrofit.Call类的扩展方法
         val response: WxArticleResponse = apiService.getWxarticle().awaitResponse()
         val sb = StringBuilder("Retrofit配合协程请求：\n")
         response.data.forEach {
             sb.append(it.name).append("\n")
         }
         tvResult.text = sb.toString()
    }
}

协程配合Retrofit（2.6.0及以上版本）发起网络请求

1. 不再需要给给Retrofit.Call类添加扩展函数(Retrofit已经给我们添加好了)。
2. 接口方法声明可以简化，方法返回类型直接声明成需要的数据类型。

interface ApiService {

    //Retrofit2.6.0以下方法声明
    //@GET("wxarticle/chapters/json")
    //fun getWxarticle(): Call<WxArticleResponse>

    //Retrofit2.6.0及以上方法声明
    @GET("wxarticle/chapters/json")
    suspend fun getWxarticle2(): WxArticleResponse

}

发起请求

private fun coroutineRequest2_6() {
    launch(exceptionHandler) {
        val response = apiService.getWxarticle2()
        val sb = StringBuilder("Retrofit2.6配合协程请求：\n")
        response.data.forEach { sb.append(it.name).append("\n") }
        tvResult.text = sb.toString()
    }
}

协程异常处理

上面的协程请求，为了捕获异常，我们声明了一个CoroutineExceptionHandler实例，然后使用。

val exceptionHandler = CoroutineExceptionHandler { _, exception ->
    Log.d(TAG, "Caught original $exception")
}

使用

private fun coroutineRequest2() {
    launch(exceptionHandler) {
        val response = apiService.getWxarticle2()
        val sb = StringBuilder("Retrofit2.6配合协程请求：\n")
        response.data.forEach { sb.append(it.name).append("\n") }
        tvResult.text = sb.toString()
    }
}

返回结果的统一处理

我们的返回结果通常有两种样式

样式1：data是一个JSONObject

{
  "data": {},
  "errorCode": 0,
  "errorMsg": ""
}

样式2：data是一个JSONArray

{
  "data": [],
  "errorCode": 0,
  "errorMsg": ""
}

首先定义通用的响应类

class NetResponse<T> {
    var data: T? = null
    var errorMsg = ""
    var errorCode = 0


    fun success() = errorCode == 0
}

Retrofit2.6.0以下格式统一处理

定义接口方法

//data是JSONObject
@GET("article/list/1/json")
fun getArticleLowLevelFormat1(): Call<NetResponse<Article>>

//data是JSONArray
@GET("wxarticle/chapters/json")
fun getWxarticleListLowLevelFormat2(): Call<NetResponse<MutableList<WxArticleResponse.DataBean>>>

使用

private fun handlerLowLevelResponseFormat1() {
    launch(exceptionHandler) {
        //需要借助Retrofit.Call类的扩展方法
        val response: NetResponse<Article> =
                apiService.getArticleLowLevelFormat1().awaitResponse()
        if (response.success()) {
            val sb = StringBuilder("Retrofit2.6以下响应格式统一处理1：\n")
            response.data?.datas?.forEach {
                sb.append(it.title).append("\n")
            }
            tvResult.text = sb.toString()
        } else {
           Log.d(TAG, "handlerLowLevelResponseFormat1: failed ${response.errorMsg}")
        }
    }
}

private fun handlerLowLevelResponseFormat2() {
    launch(handler) {
        val response: NetResponse<MutableList<WxArticleResponse.DataBean>> =
                apiService.getWxarticleListLowLevelFormat2().awaitResponse()
        if (response.success()) {
            val sb = StringBuilder("Retrofit2.6以下响应格式统一处理2：\n")
            response.data?.forEach {
                sb.append(it.name).append("\n")
                Log.d(TAG, "handlerLowLevelResponseFormat2: ${it.name}")
            }
            tvResult.text = sb.toString()
        } else {
            Log.d(TAG, "handlerLowLevelResponseFormat2: failed ${response.errorMsg}")
        }
    }
}

Retrofit2.6.0以上格式统一处理

定义接口方法

//data是JSONObject
@GET("article/list/1/json")
suspend fun getArticle(): NetResponse<Article>

//data是JSONArray
@GET("wxarticle/chapters/json")
suspend fun getWxarticleList(): NetResponse<MutableList<WxArticleResponse.DataBean>>

使用

 private fun handlerResponseFormat1() {
    launch(handler) {
        val response = apiService.getArticle()
        if (response.success()) {
            val article: Article? = apiService.getArticle().data
            val sb = StringBuilder("handlerResponseFormat1：\n")

            article?.datas?.forEach {
                sb.append(it.title).append("\n")
                Log.d(TAG, "handlerResponseFormat1: ${it.title}")
            }
            tvResult.text = sb.toString()
        } else {
            Log.d(TAG, "handlerResponseFormat1: failed ${response.errorMsg}")
        }
    }
}

private fun handlerResponseFormat2() {
    launch(handler) {
        val response = apiService.getWxarticleList()
        if (response.success()) {
            val articleList: MutableList<WxArticleResponse.DataBean>? = apiService.getWxarticleList().data
            val sb = StringBuilder("handlerResponseFormat2：\n")
            articleList?.forEach {
                sb.append(it.name)append("\n")
                Log.d(TAG, "handlerResponseFormat2: ${it.name}")
            }
            tvResult.text = sb.toString()
        } else {
            Log.d(TAG, "handlerResponseFormat2: failed ${response.errorMsg}")
        }
    }
}

参考链接：

• Android中用Kotlin Coroutine(协程)和Retrofit进行网络请求和取消请求
• Retrofit 2.6.0 ! 更快捷的协程体验 ！
• Android 开发中 Kotlin Coroutines 如何优雅地处理异常
• https://github.com/Kotlin/kotlinx.coroutines
• What is the difference between launch/join and async/await in Kotlin coroutines