3 分钟阅读

Kotlin协程的使用与封装

前言:

相信作为一个现代Android开发者,应该都用过或者听过协程了,还不少大佬都已经把协程给扒皮了,其本质就是线程池的封装 。源码的解析和性能的对比,都有解说。

协程的性能还不如原生线程池,为什么我要用协程,是因为协程可以把碎片化的方法很方便的加入异步处理,发挥Android设备多核的优势。合理的使用协程,应用反而更流畅。

下面我就不涉及太多原理理论了,直接上代码,Android中如何使用和封装协程。

一. 协程的使用

常用的几个关键的函数方法

launch ,runBlocking, withContext ,async/await

前两者启动协程,后两者调度线程。

lauch 是非阻塞的 而runBlocking是阻塞的。直接上例子:

   private fun testCoroutine1() {
       //这里只是协程作用域
      // GlobalScope.launch { 
      // lifecycleScope.launch { 
      // viewModelScope.launch { 都可以
        CoroutineScope(Dispatchers.Main).launch {
            delay(500)
            YYLogUtils.w("协程1作用域内部执行")
        }
        YYLogUtils.w("协程1作用域wai部执行")
    }

结果是先执行外部,再执行内部

而runBlocking恰恰相反

  private fun testCoroutine2() {
        runBlocking {
            delay(500)
            YYLogUtils.w("协程2作用域内部执行")
        }
        YYLogUtils.w("协程2作用域wai部执行")
    }

结果是先执行内部,再执行外部,因为阻塞了。

所以一般我们开发绝大多数都是使用launch了。 而我们切换线程一般用withContext和 async/await.区别就是你想顺序执行还是并发执行。

顺序执行: 这里会先等待1秒输入1234,然后调用接口获取Industry,请求完成之后再调用接口获取School,当前全部完成之后隐藏Loading。 其中网络请求异常的处理已经在内部封装处理了,后面会讲到。

       viewModelScope.launch {
                //开始Loading
                loadStartProgress()

                val startTimeStamp = System.currentTimeMillis()
                val res = withContext(Dispatchers.Default) {
                    //异步执行
                    delay(1000)
                    return@withContext "1234"
                }
                val endTimeStamp = System.currentTimeMillis()
                YYLogUtils.w("res: $res  time: ${endTimeStamp-startTimeStamp}")

                //网络请求获取行业数据
                val industrys = mRepository.getIndustry()

                //返回的数据是封装过的,检查是否成功
                industrys.checkResult({
                    //成功
                    _industryLD.postValue(it)
                }, {
                    //失败
                    toastError(it)
                })

                //上面的请求执行完毕才会执行这个请求
                val schools = mRepository.getSchool()
                //返回的数据是封装过的,检查是否成功
                schools.checkSuccess {
                    _schoollLD.postValue(it)
                }

                //完成Loading
                loadHideProgress()
            }

并发执行: 这里会同时调用Industry和School接口,等待两者都完成之后再展示UI。

        viewModelScope.launch {

                //开始Loading
                loadStartProgress()

                val industryResult = async {
                    mRepository.getIndustry()
                }

                val schoolResult = async {
                    mRepository.getSchool()
                }


                val localDBResult = async {
                    //loadDB()
                    YYLogUtils.w("thread:" + CommUtils.isRunOnUIThread())

                    delay(10000)
                }

                //一起处理数据
                val industry = industryResult.await()
                val school = schoolResult.await()

                //如果都成功了才一起返回
                if (industry is OkResult.Success && school is OkResult.Success) {
                    loadHideProgress()

                    _industryLD.postValue(industry.data!!)
                    _schoollLD.postValue(school.data!!)
                }


                YYLogUtils.e(localDBResult.await().toString() + "完成")

            }

大家开发App常用的两种方式都已经掌握了,还有一个不常用但是很重要的点,就是网络请求去重。 场景:点击CheckBox调用接口是否开启通知,那么我们就要把用户推送id传给服务器。如果用户狂点CheckBox,那么我怎么请求网络?

常用的两种去重手段。一种是取消上一次的,另一种是队列排队一个一来。

老规矩直接上代码了:

    /**
     * 网络请求去重
     */
    private var controlledRunner = ControlledRunner<OkResult<List<Industry>>>()  //取消之前的
    private val singleRunner = SingleRunner()       //任务队列,排队,单独的
    fun netDuplicate() {
      
        viewModelScope.launch {
            //比较常用
            //取消上一次的,执行这一次的
            controlledRunner.cancelPreviousThenRun {
                return@cancelPreviousThenRun mRepository.getIndustry()
            }.checkSuccess {
                YYLogUtils.e("请求成功:")
                _industryLD.postValue(it)
            }

            //前一个执行完毕了,再执行下一个
//                singleRunner.afterPrevious {
//                    mMainRepository.getIndustry()
//                }.checkSuccess {
//                    YYLogUtils.e("测试重复的数据:" + it.toString())
//                }

        }
      
    }

控制器源码如下:

class SingleRunner {

    private val mutex = Mutex()

    /**
     * 加入到任务队列,前一个任务执行完毕再执行下一个任务
     */
    suspend fun <T> afterPrevious(block: suspend () -> T): T {
        mutex.withLock {
            return block()
        }
    }
}

class ControlledRunner<T> {

    private val activeTask = AtomicReference<Deferred<T>?>(null)

    suspend fun cancelPreviousThenRun(block: suspend () -> T): T {

        activeTask.get()?.cancelAndJoin()

        return coroutineScope {
            val newTask = async(start = LAZY) {
                block()
            }

            newTask.invokeOnCompletion {
                activeTask.compareAndSet(newTask, null)
            }

            val result: T

            while (true) {
                if (!activeTask.compareAndSet(null, newTask)) {
                    activeTask.get()?.cancelAndJoin()
                    yield()
                } else {
                    result = newTask.await()
                    break
                }
            }

            result
        }
    }

    /**
     * 不执行新任务,返回上一个任务的结果
     */
    suspend fun joinPreviousOrRun(block: suspend () -> T): T {

        activeTask.get()?.let {
            return it.await()
        }
        return coroutineScope {

            val newTask = async(start = LAZY) {
                block()
            }

            newTask.invokeOnCompletion {
                activeTask.compareAndSet(newTask, null)
            }

            val result: T

            while (true) {
                if (!activeTask.compareAndSet(null, newTask)) {

                    val currentTask = activeTask.get()
                    if (currentTask != null) {
                        newTask.cancel()
                        result = currentTask.await()
                        break
                    } else {
                        yield()
                    }
                } else {

                    result = newTask.await()
                    break
                }
            }
            result
        }
    }
}

二. 网络请求协程的封装

Retrofit+协程的使用: 原理就是调用Retrofit方法,对它try-catch.得到的是网络请求错误信息,可以根据不同的Type类型。然后对Retrofit的返回结果再判断如果code不是200,那么就是Api错误( 例如Token失效)。对错误和成果的结果做统一的封装返回给ViewModel处理。

方式一: 处理BaseRepository:

open class BaseRepository {

    //无异常处理 -> 一般不用这个,一旦报错会App崩溃
    suspend inline fun <T : Any> handleApiCall(call: suspend () -> BaseBean<T>): BaseBean<T> {
        return call.invoke()
    }

    /**
     * 推荐使用拓展函数extRequestHttp
     * 如果要使用Base里面的方法请求网络这么使用
     *   return handleErrorApiCall(call = {
                    handleApiErrorResponse()
                })
     * 都可以实现网络请求
     */
    //处理Http错误-内部再处理Api错误
    suspend fun <T : Any> handleErrorApiCall(call: suspend () -> OkResult<T>, errorMessage: String = ""): OkResult<T> {
        return try {
            call()
        } catch (e: Exception) {
            if (!TextUtils.isEmpty(errorMessage)) {
                OkResult.Error(IOException(errorMessage))
            } else {
                OkResult.Error(handleExceptionMessage(e))
            }
        }
    }

    //处理Api错误,例如403Token过期 把BaseBean的数据转换为自定义的Result数据
    suspend fun <T : Any> handleApiErrorResponse(
        response: BaseBean<T>,
        successBlock: (suspend CoroutineScope.() -> Unit)? = null,
        errorBlock: (suspend CoroutineScope.() -> Unit)? = null
    ): OkResult<T> {

        return coroutineScope {
            //执行挂起函数
            if (response.code == 200) {  //这里根据业务逻辑来 200 -1 等
                successBlock?.let { it() }
                OkResult.Success(response.data)
            } else {
                errorBlock?.let { it() }
                OkResult.Error(IOException(response.message))
            }
        }
    }


    //处理自定义错误消息
    fun handleExceptionMessage(e: Exception): IOException {
        return when (e) {
            is UnknownHostException -> IOException("Unable to access domain name, unknown domain name.")
            is JsonParseException -> IOException("Data parsing exception.")
            is HttpException -> IOException("The server is on business. Please try again later.")
            is ConnectException -> IOException("Network connection exception, please check the network.")
            is SocketException -> IOException("Network connection exception, please check the network.")
            is SocketTimeoutException -> IOException("Network connection timeout.")
            is RuntimeException -> IOException("Error running, please try again.")
            else -> IOException("unknown error.")
        }

    }

}

使用如下:

   suspend fun getServerTime(): OkResult<ServerTimeBean> {
        return handleErrorApiCall({
            handleApiErrorResponse(
                MainRetrofit.apiService.getServerTime(
                    Constants.NETWORK_CONTENT_TYPE,
                    Constants.NETWORK_ACCEPT_V1
                )
            )
        })
    }

方式二: 使用扩展方法的直接一步到位处理:

suspend fun <T : Any> BaseRepository.extRequestHttp(call: suspend () -> BaseBean<T>): OkResult<T> {

    //两种方式都可以,自用下面一种方式
//    runCatching {
//        call.invoke()
//    }.onSuccess { response: BaseBean<T> ->
//        if (response.code == 200) {
//            OkResult.Success(response.data)
//        } else {
//            OkResult.Error(ApiException(response.code, response.message))
//        }
//    }.onFailure { e ->
//        e.printStackTrace()
//        OkResult.Error(handleExceptionMessage(Exception(e.message, e)))
//    }

    return try {

        val response = call()

        if (response.code == 200) {
            OkResult.Success(response.data)
        } else {
            OkResult.Error(ApiException(response.code, response.message))
        }

    } catch (e: Exception) {

        e.printStackTrace()
        OkResult.Error(handleExceptionMessage(e))
    }

}

使用:

   suspend inline fun getIndustry(): OkResult<List<Industry>> {

        return extRequestHttp {
            DemoRetrofit.apiService.getIndustry(
                Constants.NETWORK_CONTENT_TYPE,
                Constants.NETWORK_ACCEPT_V1
            )
        }

    }

调用接口都是固定的模板代码,和之前MVP的方式一样,只需要定义Retrofit-Api的接口定义就行。

源码 在此。

留下评论