下载带有改造和 rxJava 的文件列表

Question

我正在尝试从带有 retrofit 和 rxjava 的 api 下载文件列表，我读了很多，但我无法弄清楚 retrofit 和 rxjava 如何协同工作，我不确定观察者必须使用哪种线程

我正在尝试更改最初使用 doAsync (Anko) 制作的应用程序的文件下载过程，该应用程序在具有 api 19 且互联网连接速度非常有限的平板电脑上运行。最初的问题是平板电脑在第一个文件下载时发生内存泄漏（大约 200 个文件，使用 100 mb 内存（全部））。因此，经过长时间的研究，我决定尝试使用 Rx 和改造重写下载过程的代码（但我在 kotlin 或 movil 方面有很多经验）。

我的模特

data class File(
         val id: String
)      //staging

data class Movimientos(val borrar: List<File>,
                       val descargar: List<File>,
                       val estructura: List<Nodo>)

data class Nodo( val id: String,
                 val root: String,
                 val name: String,
                 val file: Boolean,
                 val descargado: Boolean,
                 val version: Int)

我的服务

//return a json with a list of files to download (and to more list -not important for now-)
 @GET("index.php")
    fun getMovimientos(@Query("r") r: String = "movimientos",
                       @Query("imei") id: String,
                       @Query("access-token") accesstoken: String): Observable<Movimientos>


    @Streaming
    @GET("index.php")
    fun getArchivo(@Query("r") r: String = "descarga",
                    @Query("id") fileID: String,
                    @Query("access-token") accesstoken: String): Observable<ResponseBody>


    companion object {
        fun create(): WikiApiService {

            val retrofit = Retrofit.Builder()
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                    .addConverterFactory(GsonConverterFactory.create())
                    .baseUrl("https://url/api.biblioteca/")
                    .build()

            return retrofit.create(WikiApiService::class.java)
        }
    }

调用获取列表下载并调用下载过程

private fun beginSearch(searchString: String) {
        disposable = wikiApiServe.getMovimientos("movimientos", "d55a374eebc242a5", "XXXXXXXXXXXX")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

                .subscribe(
                        {
                            result -> txt_search_result.text = "${result.estructura.size} tamaño estructura"
                            Log.d("Estructura: ", "Es: ${result.estructura}")
                            //var aux =10

                            result.estructura.forEach {
                                nodo ->
                                if (nodo.file/) {
                                    //aux--
                                    descargarArchivo(nodo.id)

                                }
                            }
                        },
                        { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }
                )




    }

调用下载函数

private fun descargarArchivo(id: String){
        disposable = wikiApiServe.getArchivo("descarga", id, "XXXXXXXXXXX")
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .subscribe(
                        {
                            //result -> txt_search_result.text = "${result.estructura.size} tamaño estructura"
                            totalDescargas = totalDescargas +1
                            Log.d("Descargado: ", "Id archivo: $id bytes: ${it.bytes().size} Total descargas: $totalDescargas")

                            //total.text = totalDescargas
                        },
                        { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }
                )

    }

当我运行代码时，应用程序会下载一些文件（大约五个），然后不再执行任何操作（logcat 向我发送了很多 GC_FOR_ALLOC 警告和跳帧）。我尝试限制对下载方法的调用（以编程方式使用计数变量（限制为 10），以避免尝试同时下载 200 个文件）并且工作但仍然发送 GC_FOR_ALLOC 并且我必须添加一些代码来下载其他 190 个文件十个批次（但我认为这不是最好的方法）。我看到我必须使用我的观察者列表并使用 flatMap 作为另一个观察者调用 downlad 函数。但我不明白如何使用这种方法。

我尝试使用另一个线程（每次下载一个新线程），但线程池出现错误

Answer 1

我会考虑将您的代码转换为使用 Observable.fromIterable 和 flatMap，flatMap 有一个带有 maxConcurrency 参数的重载，您可以使用它来避免创建 Schedulers.io() 默认情况下会执行的无限数量的线程。

在包含 foreach 循环的订阅之前，添加对 flatMap 的调用，如下所示。然后，您的 subscribe 将被调用，每次调用 getArchivo 的结果，因此用您当前在 getArchivo subscribe 中拥有的代码替换它的内容

private fun beginSearch(searchString: String) {
    disposable = wikiApiServe.getMovimientos("movimientos", "d55a374eebc242a5", "XXXXXXXXXXXX")
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .flatMap { result ->
            Observable.fromIterable(result.estructura)
                .flatMap({ nodo ->
                    wikiApiServe.getArchivo("descarga", nodo.id, "XXXXXXXXXXX")
                    .subscribeOn(Schedulers.io())
            }, 4)
        }
        .subscribe(
            {
            totalDescargas = totalDescargas +1
            Log.d("Descargado: ", "Id archivo: $id bytes: ${it.bytes().size} Total descargas: $totalDescargas")
            },
            { error -> Toast.makeText(this, error.message, Toast.LENGTH_SHORT).show() }