Soy nuevo en Kotlin/Coroutines, así que espero que me esté perdiendo algo/no entiendo completamente cómo estructurar mi código para el problema que estoy tratando de resolver.
Esencialmente, estoy tomando una lista de cadenas, y para cada elemento de la lista quiero enviarlo a otro método para que funcione (haga una llamada de red y devuelva datos según la respuesta). ( Editar :) Quiero que todas las llamadas se inicien al mismo tiempo y se bloqueen hasta que todas las llamadas se realicen/se actúe sobre la respuesta, y luego devuelva una nueva lista con la información de cada respuesta.
Probablemente todavía no entiendo completamente cuándo usar launch/async, pero he intentado seguir tanto con launch (con joinAll ) como con async (con await ).
fun processData(lstInputs: List<String>): List<response> { val lstOfReturnData = mutableListOf<response>() runBlocking { withContext(Dispatchers.IO) { val jobs = List(lstInputs.size) { launch { lstOfReturnData.add(networkCallToGetData(lstInputs[it])) } } jobs.joinAll() } } return lstofReturnData Lo que espero que suceda es que si mi lstInputs tiene un tamaño de 120, cuando se unen todos los trabajos, mi lstOfReturnData también debería tener un tamaño de 120.
Lo que realmente está sucediendo son resultados inconsistentes. Lo ejecutaré una vez y obtengo 118 en mi lista final, lo ejecuto de nuevo, es 120, lo ejecuto de nuevo, es 117, etc. En el método networkCallToGetData() , estoy manejando cualquier excepción, para al menos devolver algo para cada solicitud, independientemente de si la llamada de red falla.
¿Alguien puede ayudarme a explicar por qué obtengo resultados inconsistentes y qué debo hacer para asegurarme de que estoy bloqueando adecuadamente y que todos los trabajos se unen antes de continuar?
mutableListOf() crea una ArrayList , que no es segura para subprocesos.
Intente usar ConcurrentLinkedQueue en su lugar.
Además, ¿usáis la versión estable de Kotlin/Kotlinx.coroutine (no la anterior experimental)? En la versión estable, con la introducción de la concurrencia estructurada, ya no es necesario escribir jobs.joinAll anymore . launch es una función de extensión de runBlocking que lanzará nuevas rutinas en el alcance de runBlocking y el alcance de runBlocking esperará automáticamente a que finalicen todos los trabajos lanzados. Así que el código anterior se puede acortar a
val lstOfReturnData = ConcurrentLinkedQueue<response>() runBlocking { lstInputs.forEach { launch(Dispatches.IO) { lstOfReturnData.add(networkCallToGetData(it)) } } } return lstOfReturnDatarunBlocking bloquea el subproceso actual de forma interrumpible hasta su finalización. Supongo que no es lo que quieres. Si pienso mal y desea bloquear el hilo actual, puede deshacerse de la rutina y simplemente hacer una llamada de red en el hilo actual:
val lstOfReturnData = mutableListOf<response>() lstInputs.forEach { lstOfReturnData.add(networkCallToGetData(it)) }Pero si no es su intención, puede hacer lo siguiente:
class Presenter(private val uiContext: CoroutineContext = Dispatchers.Main) : CoroutineScope { // creating local scope for coroutines private var job: Job = Job() override val coroutineContext: CoroutineContext get() = uiContext + job // call this to cancel job when you don't need it anymore fun detach() { job.cancel() } fun processData(lstInputs: List<String>) { launch { val deferredList = lstInputs.map { async(Dispatchers.IO) { networkCallToGetData(it) } // runs in parallel in background thread } val lstOfReturnData = deferredList.awaitAll() // waiting while all requests are finished without blocking the current thread // use lstOfReturnData in Main Thread, eg update UI } } }Runblocking debería significar que no tiene que llamar a unirse. Lanzar una corrutina desde dentro de un alcance de bloqueo de ejecución debería hacer esto por usted. ¿Has probado solo:
fun processData(lstInputs: List<String>): List<response> { val lstOfReturnData = mutableListOf<response>() runBlocking { lstInputs.forEach { launch(Dispatchers.IO) { lstOfReturnData.add(networkCallToGetData(it)) } } } return lstofReturnData