《Kotlin系列》之协程+Flow+Okhttp3实现多任务下载（暂停、继续）

先看看效果图

封装

• 1、下载器:DownloadManager

class DownloadManager(context: Context?) {


    private var completeInfo: MutableMap<String, DownloadInfo>? = HashMap()
    private var downloadInfo: MutableMap<String, DownloadInfo>? = HashMap()

    private var client: OkHttpClient? = null

    private var context: Context? = null

    private var maxRequests = 5 //最大并发数

    companion object {


        const val DOWNLOAD_STATE_WAITING = 0x00 //等待

        const val DOWNLOAD_STATE_DOWNLOADING = 0x01 //下载中

        const val DOWNLOAD_STATE_PAUSE = 0x02 //暂停

        const val DOWNLOAD_STATE_CANCLE = 0x03 //取消

        const val DOWNLOAD_STATE_FINISH = 0x04 //完成

        const val DOWNLOAD_STATE_FAIL = 0x05 //失败

        const val DOWNLOAD_STATE_RESTART = 0x06 //重新下载


        const val DOWNLOAD_MAPS = "DOWNLOAD_MAPS" //下载队列的
        const val COMPLETE_MAPS = "COMPLETE_MAPS" //已完成的

        @SuppressLint("StaticFieldLeak")
        private var instance: DownloadManager? = null


        fun get(context: Context?): DownloadManager {
            if (instance == null) {
                synchronized(DownloadManager::class.java) {
                    if (instance == null) {
                        instance = DownloadManager(context)
                    }
                }
            }
            return instance!!
        }
    }

    init {
        init(context)
    }


    /**
     * 初始化一些配置
     */
    private fun init(context: Context?) {
        client = OkHttpClient()
        client?.dispatcher()?.maxRequests = maxRequests
        this.context = context


        downloadInfo = PrefsUtil.getInstance()?.getMap(DOWNLOAD_MAPS)

        if (downloadInfo == null) {
            downloadInfo = HashMap()
        }

        completeInfo = PrefsUtil.getInstance()?.getMap(COMPLETE_MAPS)
        if (completeInfo == null) {
            completeInfo = HashMap()
        }

    }

    /**
     *保存当前任务
     */
    fun saveDownloadInfo(key: String?, downloadInfo: MutableMap<String, DownloadInfo>?) {
        key?.let { PrefsUtil.getInstance()?.putMap(it, downloadInfo) }
    }


    /**
     * 设置最大并发
     */
    fun setMaxRequests(maxRequests: Int): DownloadManager {
        this.maxRequests = maxRequests
        return this
    }


    /**
     * 取消全部下载
     */
    fun cancelAll() {
        if (client != null) {
            for (call in client?.dispatcher()?.queuedCalls()!!) {
                cancel(call.request().tag().toString())
            }
            for (call in client?.dispatcher()?.runningCalls()!!) {
                cancel(call.request().tag().toString())
            }
        }
    }

    /**
     * 取消下载
     *
     * @param url
     */
    fun cancel(url: String?) {
        if (client != null) {
            for (call in client?.dispatcher()?.queuedCalls()!!) {
                if (call.request().tag() == url) call.cancel()
            }
            for (call in client?.dispatcher()?.runningCalls()!!) {
                if (call.request().tag() == url) call.cancel()
            }
        }
        if (downloadInfo?.get(url) != null) {
            val cancelInfo: DownloadInfo? = downloadInfo?.get(url)
            cancelInfo?.setDownloadState(DOWNLOAD_STATE_CANCLE)
            downloadInfo?.remove(cancelInfo?.getUrl())
            saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
            val file = File(cancelInfo?.getTargetUrl().toString())
            if (file.exists()) file.delete()
        }
    }

    /**
     * 暂停全部下载
     */
    fun pauseAll() {
        if (client != null) {
            for (call in client?.dispatcher()?.queuedCalls()!!) {
                pause(call.request().tag().toString())
            }
            for (call in client?.dispatcher()?.runningCalls()!!) {
                pause(call.request().tag().toString())
            }
        }
    }

    /**
     * 暂停下载
     *
     * @param url
     */
    fun pause(url: String?) {
        if (client != null) {
            for (call in client?.dispatcher()?.queuedCalls()!!) {
                if (call.request().tag() == url) call.cancel()
            }
            for (call in client?.dispatcher()?.runningCalls()!!) {
                if (call.request().tag() == url) call.cancel()
            }
        }
        if (downloadInfo?.get(url) != null) {
            val pauseInfo: DownloadInfo? = downloadInfo?.get(url)
            pauseInfo?.setDownloadState(DOWNLOAD_STATE_PAUSE)
            pauseInfo?.getUrl()?.let { downloadInfo?.put(it, pauseInfo) }
            saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
        }
    }


    /**
     * 添加下载任务
     *
     * @param url                     下载请求的网址
     * @param targetUrl               下载保存的位置
     * @param callBack 用来回调的接口

     */
    fun download(
        url: String?,
        targetUrl: String?,
        callBack: DownloaderCallBack?,
        tag: Int
    ): DownloadManager {
        val downloadInfo = DownloadInfo(url, tag)
        downloadInfo.setTargetUrl(targetUrl)
        download(downloadInfo, callBack, tag)

        return this
    }


    /**
     * 添加下载任务
     *
     * @param mDownloadInfo            下载类
     * @param callBack 用来回调的接口
     */
    private fun download(
        mDownloadInfo: DownloadInfo?,
        callBack: DownloaderCallBack?,
        tag: Int?
    ) {
        if (client != null) { //包含下载url,不做处理
            for (call in client?.dispatcher()?.queuedCalls()!!) {
                if (call.request().tag() == mDownloadInfo?.getUrl()) return
            }
            for (call in client?.dispatcher()?.runningCalls()!!) {
                if (call.request().tag() == mDownloadInfo?.getUrl()) return
            }
        }
        val info: DownloadInfo?
        val request: Request?
        if (downloadInfo?.get(mDownloadInfo?.getUrl()) != null) { //在下载队列中
            info = downloadInfo?.get(mDownloadInfo?.getUrl())
            val file = info?.getTargetUrl()?.let { File(it) }
            var isNormal = true
            if (file?.exists() == true) {
                //找到了文件,代表已经下载过,则获取其长度
                info.setProgress(file.length())
                if (info.getProgress()?.let { it >= info.getTotal()!! } == true) {
                    isNormal = false
                    file.delete()
                    info.setProgress(0)
                    info.setTotal(0)
                } else callBack?.onProgress(info.getProgress(), info.getTotal(), tag)
            }
            info?.setDownloadState(DOWNLOAD_STATE_WAITING)
            info?.let { downloadInfo?.put(info.getUrl().toString(), it) }
            saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)

            request = if (isNormal) Request.Builder()
                .addHeader(
                    "RANGE",
                    "bytes=" + info?.getProgress().toString() + "-" + info?.getTotal()
                )
                .url(info?.getUrl().toString())
                .tag(info?.getUrl())
                .build() else Request.Builder()
                .url(info?.getUrl().toString())
                .tag(info?.getUrl())
                .build()
        } else { //添加新任务
            info = mDownloadInfo
            info?.setDownloadState(DOWNLOAD_STATE_WAITING)
            if (TextUtils.isEmpty(mDownloadInfo?.getFileName())) {
                var fileName = mDownloadInfo?.getTargetUrl()?.substring(
                    mDownloadInfo.getTargetUrl()!!.lastIndexOf("/")
                )
                if (fileName?.startsWith("/") == true && fileName.length > 1) fileName =
                    fileName.substring(1)
                info?.setFileName(fileName)
            }
            info?.let { downloadInfo?.put(info.getUrl().toString(), it) }
            saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)

            request = Request.Builder()
                .url(info?.getUrl().toString())
                .tag(info?.getUrl())
                .build()
        }
        val call: Call? = request?.let {
            client?.newBuilder()?.addNetworkInterceptor { chain ->

                //设置拦截器
                val originalResponse = chain.proceed(chain.request())
                originalResponse.newBuilder()
                    .body(
                        ResponseProgressBody(
                            context,
                            originalResponse.body(),
                            callBack,
                            info
                        )
                    )
                    .build()
            }
                ?.build()
                ?.newCall(it)
        }
        call?.enqueue(
            MyDownloadCallback(callBack, info)
        )
    }


    inner class MyDownloadCallback(
        val callBack: DownloaderCallBack?,
        private val info: DownloadInfo?

    ) : Callback {
        private var targetUrl: String? = info?.getTargetUrl()
        override fun onFailure(call: Call, e: IOException) {

            runBlocking {
                flow { emit(1) }.collect {
                    callBack?.onFailure(e.toString())
                    info?.setDownloadState(DOWNLOAD_STATE_FAIL)
                    info?.let { it1 -> downloadInfo?.put(info.getUrl().toString(), it1) }
                    saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
                }
            }

        }

        @OptIn(DelicateCoroutinesApi::class)
        override fun onResponse(call: Call, response: Response) {
            //第一种写法
            runBlocking {
                if (response.isSuccessful) {
                    flow {
                        emit(response)
                    }.map {
                        BannerUtils.saveFile(response, targetUrl)
                    }.catch {

                        when (info?.getDownloadState()) {

                            DOWNLOAD_STATE_CANCLE -> callBack?.onCancel(info)
                            DOWNLOAD_STATE_PAUSE -> callBack?.onPause(info)
                            else -> {
                                callBack?.onFailure("onResponse saveFile fail." + it.message)
                                info?.setDownloadState(DOWNLOAD_STATE_FAIL)
                                downloadInfo?.put(info?.getUrl()!!, info)
                                saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
                            }
                        }

                    }.collect {

                        callBack?.onFinish(it, info?.getTag())

                        info?.setDownloadState(DOWNLOAD_STATE_FINISH)
                        downloadInfo?.remove(info?.getUrl())
                        info?.let { it1 -> completeInfo?.put(info.getUrl().toString(), it1) }
                        saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
                        saveDownloadInfo(COMPLETE_MAPS, completeInfo)
                    }

                } else {
                    flow { emit(1) }
                        .collect {
                            callBack?.onFailure("fail status=" + response.code())
                            info?.setDownloadState(DOWNLOAD_STATE_FAIL)
                            info?.let { it1 -> downloadInfo?.put(info.getUrl().toString(), it1) }
                            saveDownloadInfo(DOWNLOAD_MAPS, downloadInfo)
                        }

                }

            }


        }
    }


}

• 2、下载信息类:DownloadInfo

class DownloadInfo(//下载路径
    private var url: String?,// 下載標識
    private var tag: Int
) : Serializable {
    private var targetUrl: String? = null//存储路径
    private var total: Long? = 0//总大小
    private var progress: Long? = 0 //当前进度

    private var fileName: String? = null //名称

    private var downloadState: Int? = 0 //下载状态


    fun getTag(): Int {
        return tag
    }

    fun setTag(tag: Int) {
        this.tag = tag
    }

    fun getUrl(): String? {
        return url
    }

    fun getFileName(): String? {
        return fileName
    }

    fun setFileName(fileName: String?) {
        this.fileName = fileName
    }

    fun getTotal(): Long? {
        return total
    }

    fun setTotal(total: Long?) {
        this.total = total
    }

    fun getProgress(): Long? {
        return progress
    }

    fun setProgress(progress: Long?) {
        this.progress = progress
    }

    fun getTargetUrl(): String? {
        return targetUrl
    }

    fun setTargetUrl(targetUrl: String?) {
        this.targetUrl = targetUrl
    }

    fun setDownloadState(downloadState: Int?) {
        this.downloadState = downloadState
    }

    fun getDownloadState(): Int? {
        return downloadState
    }

• 3、下载回调管理:DownloaderCallBack

abstract class DownloaderCallBack {

    abstract fun onProgress(currentBytes: Long?, totalBytes: Long?, tag: Int?)
    abstract fun onFinish(download_file: File?, tag: Int?)
    abstract fun onPause(downloadInfo: DownloadInfo?)
    abstract fun onCancel(downloadInfo: DownloadInfo?)
    abstract fun onFailure(error_msg: String?)
}

• 4、下载Body拦截:ResponseProgressBody

class ResponseProgressBody(
    private var mContext: Context?,
    private var mResponseBody: ResponseBody?,
    private var callBack: DownloaderCallBack?,
    private var info: DownloadInfo?
) : ResponseBody() {

    private var bufferedSource: BufferedSource? = null

    private var progress: Long = 0//开始前已下载进度

    private var downloadInfo: MutableMap<String?, DownloadInfo?>? = null


    init {
        progress = info?.getProgress()!!
        downloadInfo = PrefsUtil.getInstance()?.getMap(DownloadManager.DOWNLOAD_MAPS)
        if (info?.getTotal()!! <= 0) {
            info?.setTotal(mResponseBody?.contentLength())
            downloadInfo?.put(info?.getUrl(), info)
            PrefsUtil.getInstance()?.putMap(DownloadManager.DOWNLOAD_MAPS, downloadInfo)

        }
    }

    override fun contentType(): MediaType? {
        return mResponseBody?.contentType()
    }

    override fun contentLength(): Long {
        return mResponseBody?.contentLength()!!
    }

    override fun source(): BufferedSource {
        if (bufferedSource == null) {
            bufferedSource =
                mResponseBody?.source()?.let { source(it).let { it1 -> Okio.buffer(it1) } }
        }
        return bufferedSource!!
    }

    private fun source(source: Source): Source {
        return object : ForwardingSource(source) {
            var totalBytesRead: Long = 0

            @Throws(IOException::class)
            override fun read(sink: Buffer, byteCount: Long): Long {
                val bytesRead = super.read(sink, byteCount)
                totalBytesRead += if (bytesRead != -1L) bytesRead else 0
                info?.setProgress(totalBytesRead + progress)

                runBlocking {
                    flow { emit(1) }
                        .collect {
                            callBack?.onProgress(
                                info?.getProgress(),
                                info?.getTotal(),
                                info?.getTag()
                            )
                        }
                }


                info?.setDownloadState(DownloadManager.DOWNLOAD_STATE_DOWNLOADING)
                downloadInfo!![info?.getUrl()] = info
                PrefsUtil.getInstance()?.putMap(DownloadManager.DOWNLOAD_MAPS, downloadInfo)

                return bytesRead
            }
        }
    }

使用方式

• 1、初始化下载器并设置最大并发量

   downloadManager = DownloadManager.get(getApplication()).setMaxRequests(3)

• 2、下载对应url

   val url: String?
        val tag: Int?
        val name: String?
        when (index) {
            1 -> {
                url = url1?.get()
                tag = 1
                name = "origen.zip"
            }
            2 -> {
                url = url2?.get()
                tag = 2
                name = "opencv.zip"
            }
            3 -> {
                url = url3?.get()
                tag = 3
                name = "three.jpg"
            }
            else -> {
                url = url1?.get()
                tag = 1
                name = "three.jpg"
            }
        }
      downloadManager?.download(
            url,
            BannerUtils.getSdPath(getApplication(), name),
            CallBack(),
            tag
        )

• 3、回调中处理progress

   inner class CallBack : DownloaderCallBack() {
        override fun onProgress(currentBytes: Long?, totalBytes: Long?, tag: Int?) {
            totalBytes?.let {
                val sum = currentBytes?.times(1f.div(totalBytes * 1f))
                val process = sum?.times(100) ?: 0


                when (tag) {
                    1 -> process1?.set(process.toInt())
                    2 -> process2?.set(process.toInt())
                    3 -> process3?.set(process.toInt())
                    else -> process1?.set(process.toInt())
                }

                Log.e(TAG, "the tag ==$tag+ sum ==$process")
            }


        }

        override fun onFinish(download_file: File?, tag: Int?) {
            Log.e(TAG, "download finish file path=" + download_file?.path)
        }

        override fun onPause(downloadInfo: DownloadInfo?) {
            Log.e(TAG, "download status is pause" + Thread.currentThread())
        }

        override fun onCancel(downloadInfo: DownloadInfo?) {
            Log.e(TAG, "download status is cancel" + Thread.currentThread())
            when (downloadInfo?.getTag()) {
                1 -> process1?.set(0)
                2 -> process2?.set(0)
                3 -> process3?.set(0)
                else -> process1?.set(0)
            }

        }

        override fun onFailure(error_msg: String?) {
            Log.e(TAG, "download status is failure" + Thread.currentThread())
        }
    }

• 4、暂停和取消任务

   /**
     *暂停某个任务
     */
    fun pauseDownload(index: Int?) {

        val url: String? = when (index) {
            1 -> url1?.get()
            2 -> url2?.get()
            3 -> url3?.get()
            else -> url1?.get()
        }
        downloadManager?.pause(url)
    }

    /**
     * 暂停某个下载
     */
    fun cancelDownload(index: Int?) {
        val url: String? = when (index) {
            1 -> url1?.get()
            2 -> url2?.get()
            3 -> url3?.get()
            else -> url1?.get()
        }
        downloadManager?.cancel(url)
    }

代码已上传github
https://github.com/ljlstudio/KtMvvm/tree/master/demo/src/main/java/com/kt/ktmvvm/download

最后

以上就是迷你夏天为你收集整理的《Kotlin系列》之协程+Flow+Okhttp3实现多任务下载（暂停、继续）的全部内容，希望文章能够帮你解决《Kotlin系列》之协程+Flow+Okhttp3实现多任务下载（暂停、继续）所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。