使用 Coroutines 读取和复制文件

Question

我创建了以下应用程序来说明一些疑问。 My Example on the Github

在本例中，我将一个文件复制到另一个包中。

我的疑惑如下：

1. 并行执行任务，是否可以返回取消前完成的值？

2. 为什么在contentResolver.openInputStream (uri) 中出现消息“不适当的阻塞方法调用”，而我正在使用 IO 上下文？

3. 当我读取文件条目复制到输出时，我总是检查作业状态，以便当这个任务被取消时，它立即停止，创建的输出文件被删除并返回取消异常，是对吗？

4. 我可以限定执行的任务数量吗？

我的 onCreate：

private val listUri = mutableListOf<Uri>()
private val job = Job()

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    setContentView(R.layout.activity_main)
    
    //get files from 1 to 40
    val packageName = "android.resource://${packageName}/raw/"
    for (i in 1..40) {
        listUri.add(Uri.parse("${packageName}file$i"))
    }
}

我的按钮操作：

  //Button action
   fun onClickStartTask(view: View) {
        var listNewPath = emptyList<String>()
        CoroutineScope(Main + job).launch {
            try {
                //shows something in the UI - progressBar
                withContext(IO) {
                    listNewPath = listUri.map { uri ->
                        async {
                            //path to file temp
                            val pathFileTemp =
                                "${getExternalFilesDir("Temp").toString()}/${uri.lastPathSegment}"
                            val file = File(pathFileTemp)
                            val inputStream = contentResolver.openInputStream(uri)
                            inputStream?.use { input ->
                                FileOutputStream(file).use { output ->
                                    val buffer = ByteArray(1024)
                                    var read: Int = input.read(buffer)
                                    while (read != -1) {
                                        if (isActive) {
                                            output.write(buffer, 0, read)
                                            read = input.read(buffer)
                                        } else {
                                            input.close()
                                            output.close()
                                            file.deleteRecursively()
                                            throw CancellationException()
                                        }
                                    }
                                }
                            }
                            //If completed then it returns the new path.
                            return@async pathFileTemp
                        }
                    }.awaitAll()
                }
            } finally {
                //shows list complete in the UI
            }
        }
    }

我的取消作业按钮：

fun onClickCancelTask(view: View) {
    if (job.isActive) {
        job.cancelChildren()
        println("Cancel children")
    }
}

这将是执行任务的按钮操作。

感谢大家的帮助。

Answer 1

回答 1. 和 4.：

为了划分并行任务并让它们独立完成（获取一些值，同时取消其余的），您需要使用 Channel，最好使用 Flow。简化示例：

fun processListWithSomeWorkers(list: List<Whatever>, concurrency: Int): Flow<Result> = channelFlow {
   val workToDistribute = Channel<Whatever>()
   launch { for(item in list) workToDistribute.send(item) } // one coroutine distributes work...

    repeat(concurrency) { // launch a specified number of worker coroutines
      launch { 
         for (task in workToDistribute) { // which process tasks in a loop
            val atomicResult = process(task)
            send(atomicResult) // and send results downstream to a Flow
         }
      }
   }
}

然后您可以一个一个地处理结果，因为它们正在等待整个流程完成或例如只需在需要时取其中一些： resultFlow.take(20).onEach { ... }.collectIn(someScope) 因为它是一个 Flow，它只有在有人开始收集时才会开始工作（天气很冷），这通常是一件好事。

整个事情可能会更短一些，因为您会发现一些更具体和实验性的功能（作为产品）。它可以概括为这样的 Flow 运算符：

fun <T, R> Flow<T>.concurrentMap(concurrency: Int, transform: suspend (T) -> R): Flow<R> {
    require(concurrency > 1) { "No sense with concurrency < 2" }
    return channelFlow {
        val inputChannel = produceIn(this)
        repeat(concurrency) {
            launch {
                for (input in inputChannel) send(transform(input))
            }
        }
    }
}

并使用：list.asFlow().concurrentMap(concurrency = 4) { <your mapping logic> }

corotuines 团队正在考虑将一系列并行运算符添加到 Flow 流中，但它们还没有 AFAIK。

Answer 2

我认为这是一个更好的方法

fun onClickStartTask(view: View) {
    var listNewPath = emptyList<String>()
    val copiedFiles = mutableListOf<File>()
    CoroutineScope(Dispatchers.Main + job).launch {
        try {
            //shows something in the UI - progressBar
            withContext(Dispatchers.IO) {
                listNewPath = listUri.map { uri ->
                    async {
                        //path to file temp
                        val pathFileTemp =
                                "${getExternalFilesDir("Temp").toString()}/${uri.lastPathSegment}"
                        val file = File(pathFileTemp)
                        val inputStream = contentResolver.openInputStream(uri)
                        inputStream?.use { input ->
                            file.outputStream().use { output ->
                                copiedFiles.add(file)
                                input.copyTo(output, 1024)
                            }
                        }

                        //If completed then it returns the new path.
                        return@async pathFileTemp
                    }
                }.awaitAll()
            }
        } finally {
            //shows list complete in the UI
        }
    }
    job.invokeOnCompletion {
        it?.takeIf { it is CancellationException }?.let {
            GlobalScope.launch(Dispatchers.IO) {
                copiedFiles.forEach { file ->
                    file.delete()
                }
            }
        }
    }
}