基於 Flow 的多線程並行通用圖片上傳框架

基於 Flow 的多線程並行通用圖片上傳框架，結合 LiveData 監聽回調

之前年少無知寫過一篇文章 利用策略模式結合alibaba/alpha框架優化你的圖片上傳功能，我以爲還能夠搶救一下啊，因此如今將這個項目重構了一下變成如下所說的。

說到上傳下載，你們確定有各類方法，這裏主要是利用了 Kotlin 的 Flow 多線程去實現，而且是並行且最後結果按順序收集。

地址 Github

特色：

1. 對外開放上傳接口，具體上傳邏輯本身定義
2. 基於 Kotlin 的 Flow 實現多線程並行上傳，而且上傳結果按順序返回哦
3. 結合 LiveData 監聽上傳回調，管理生命週期
4. 提供攔截器功能，攔截器可在主線程和子線程中使用，提供過濾器功能
5. 代碼簡單容易擴展和修改
6. 其實上傳也能夠用哦

API 預覽：

MultipleUpload
    .with( .. )    //傳入當前的 Lifecycle
    .load( .. )    //傳入上傳的路徑，可傳一個，可傳多個
    .setUploadImpl( .. )    //配置本身具體的上傳邏輯
    .setThreadPoolExecutor( .. )     //配置自定義的線程池，不然用默認的
    .filter { .. }         // 路徑過濾器，dsl 形式
    .filter(object : UploadFilter { ..  })   // 路徑過濾器，接口形式
    .addInterceptor(object : UploadIntercept() { .. }, UploadIntercept.UI) //添加攔截器，第二個參數是攔截器運行的線程。不傳默認是UI
    .addInterceptor(object : UploadIntercept() { .. }, UploadIntercept.IO) //添加攔截器，運行在子線程
    .singleUploadObserver {        //單個上傳的監聽回調，dsl 形式
        onStart { index-> .. }     //開始上傳，index 是當前上傳的第幾個文件
        onProgress { index, progress, totalProgress -> .. } //上傳進度
        onSuccess { index, url, otherParams -> ..  }        //上傳成功
        onFailure { index, errCode, errStr -> .. }          //上傳失敗
    }
    .singleUploadObserver(object : OnSingleUploadState{ .. })  //單個上傳的監聽回調，接口形式
    .multipleUploadObserver {         //多個文件上傳時整體監聽回調，dsl 形式
        onStart { .. }                //上傳開始
        onCompletion { successNum, failNum, urls -> .. }   //所有上傳完成
        onFailure { catchIndex, errStr -> .. }             //上傳中發生 catch。
    }
    .multipleUploadObserver(object : OnMultipleUploadState{ .. }) //多個文件上傳時整體監聽回調，接口形式
    .upload()  //發起上傳
複製代碼

API詳解

1. with

fun with(context: Context)
fun with(activity: Activity)
fun with(context: FragmentActivity)
fun with(fragment: Fragment)
fun with(owner: LifecycleOwner = ProcessLifecycleOwner.get())
複製代碼

with 方法有多個重載，能夠傳入 context，activity 等，但 context 和 activity 都必須是 FragmentActivity 的子類，由於最終須要的是 LifecycleOwner。 若是不傳，則默認是 ProcessLifecycleOwner.get()。

2. load

fun load(path: String?)
fun load(list: MutableList<String?>)
fun load(path: String?, params: HashMap<String, Any>)
fun load(list: MutableList<String?>, params: HashMap<String, Any>)
複製代碼

load 方法有 4 個重載，可分紅兩類，一類是傳入一個路徑，即上傳一個文件，一類是傳入一個 List，即同時上傳多個文件。

params 參數的意思是，在咱們實現上傳邏輯的時候，有時候不僅須要一個 path，可能還須要其餘一些業務參數，這時候就能夠用到 params 來傳遞了。

3. setUploadImpl

fun setUploadImpl(upload: UploadInterface)

interface UploadInterface {
    fun uploadFile(path: String, params: HashMap<String, Any>, callback: UploadCallback)
}

interface UploadCallback {
    fun onUploadStart()
    fun onUploadProgress(progress: Int, totalProgress: Int)
    fun onUploadSuccess(url: String, otherParams: HashMap<String, Any>? = null)
    fun onUploadFail(errCode: Int, errMsg: String?)
}
複製代碼

傳入具體的上傳邏輯，參數是 UploadInterface 接口，經過實現這個接口來實現本身具體的上傳邏輯，而後經過 callback 回調給框架處理。 能夠看到 uploadFile 方法的第二個參數就是上面 load 方法中說到的那個自定義參數了。

同時能夠看到 onUploadSuccess 回調中第二個參數 otherParams，也是由於上傳完成後可能須要傳遞一些其餘業務內容出去，這時候也能夠用到它去作。

4. setThreadPoolExecutor

配置自定義的線程池，若是你想上傳時的線程池由本身去作，能夠經過它來傳進去，參數是 ThreadPoolExecutor，不傳的話會使用默認的。

5. filter

fun filter(filter: UploadFilter)
fun filter(filter: (String) -> Boolean)

interface UploadFilter {
    fun apply(path: String): Boolean
}
複製代碼

filter 方法的做用是作一些路徑過濾，須要實現 UploadFilter 接口，會過濾掉返回 false 的路徑，它的調用時機是在上傳前。

有兩個重載，分別是接口形式和 dsl形式，能夠理解爲一個用於 kotlin，一個用於兼容 java。

隨便一提，像判空，判斷本地是否存在改文件這種過濾內部已經有了，就不須要本身去作了。

6. addInterceptor

fun addInterceptor(interceptor: UploadIntercept,interceptThread: String = UploadIntercept.UI)

abstract class UploadIntercept {
    companion object {
        const val UI = "UI"
        const val IO = "IO"
    }

    open fun processSingle(path: String, callback: InterceptCallback) {}  //用於單個文件
    open fun processMultiple(paths: MutableList<String>, callback: InterceptCallback) {} //用於多個文件
}

interface InterceptCallback {
    fun onNext(path: String) //執行下一個，用於上傳一個文件
    fun onNext(paths: MutableList<String>) //執行下一個，用於上傳多個文件
    fun onInterrupt(code: Int = -1, msg: String?) //中斷 code，msg:能夠添加code 和 msg，若是 msg 不爲空，則會回調失敗回調
}
複製代碼

攔截器，做用你們應該很熟悉了，運行時機是在 filter 以後，上傳以前。

實現攔截器需求實現 UploadIntercept，由於上傳單個文件就一個路徑，多個文件有多個路徑，是一個 List，因此爲了區分，就有了 processSingle 和 processMultiple 兩個方法，能夠根據須要實現。

InterceptCallback 回調接口， onNext 方法表明執行下一個攔截邏輯，一樣分兩個，做用也跟剛剛說的同樣。

若是要中斷邏輯，則能夠調用 onInterrupt 方法，有兩個參數，若是 msg 有值的話，會回調上傳失敗。

addInterceptor 添加攔截器的第二個參數 interceptThread，表明攔截器運行在 UI 仍是 IO 線程，默認是 UI 線程。

7. singleUploadObserver

fun singleUploadObserver(observer: SingleUploadObserver.() -> Unit)
fun singleUploadObserver(state: OnSingleUploadState)
複製代碼

單個文件上傳監聽，一樣有兩個重載，分別是接口形式和 dsl形式，能夠理解爲一個用於 kotlin，一個用於兼容 java。它是基於 LiveData 的。

有幾個回調分別是 onStart，onProgress，onSuccess 和 onFailure。singleUploadObserver 無輪是上傳多個仍是一個文件的時候都會有回調。

它們都有一個參數 index 表明着目前正在上傳第幾個文件。onSuccess 中有一個參數是 otherParams，它的做用是上面第 3 點中 setUploadImpl 說到的。

8. multipleUploadObserver

fun multipleUploadObserver(observer: MultipleUploadObserver.() -> Unit)
fun multipleUploadObserver(state: OnMultipleUploadState)
複製代碼

在多個文件上傳時，除了須要瞭解每一個文件上傳的狀況（即上面的 singleUploadObserver），還須要瞭解一些整體狀況，好比上傳開始，所有上傳結束等， multipleUploadObserver 就是這個做用，一樣兩個重載，接口形式和 dsl形式，用於 kotlin，一個用於兼容 java。

有三個回調是 onStart，onCompletion，和 onFailure，表明上傳開始，所有上傳結束和上傳發生錯誤。

在 onCompletion 中，你能夠拿到 successNum（上傳文件成功數），failNum（上傳文件失敗數），urls（上傳完成後文件路徑集合，注意這個集合的順序是跟你傳入上傳路徑那個集合順序是同樣的 ）

9. upload

發起上傳邏輯，調用這個方法後纔會真正發起上傳邏輯。

Flow 多線程並行而且按順序返回原理：

scope?.launch {
    uploadList.mapIndexed { index, path ->
        scope.async(supportDispatcher) {
            uploadImpl()
        }
    }.asFlow().map {
        it.await()
    }.flowOn(supportDispatcher)
}
複製代碼

是的，就是這麼幾行代碼。

運行效果

示例代碼你們能夠在 MainActivity 中找到，這裏模擬上傳，每一個文件上傳須要三秒。

1. 單個文件上傳效果

2. 3個文件同時上傳效果