Consultas paralelas en Kotlin para automatizar la recopilación de datos

¡Hola a todos! En mi trabajo, suelo utilizar Kotlin para la automatización. Mi actividad no está directamente relacionada con la programación, pero Kotlin simplifica enormemente las tareas laborales.

Recientemente fue necesario recolectar datos de un tamaño bastante grande para poder hacer el análisis, así que decidí escribir un pequeño script para obtener los datos y guardarlos en Excel. No hubo problemas con el último punto: leí sobre Apache POI, tomé un par de ejemplos de la documentación oficial y los modifiqué por mí mismo. No se puede decir lo mismo de las solicitudes de Internet.

La fuente regresó en lotes de json y fue necesario recopilar estos "lotes" de alguna manera rápidamente, convertirlos en texto y escribir una tabla en un archivo.

Método asincrónico

Decidí comenzar con una simple asincronización. Después de hurgar un poco en HttpUrlConnection, lo envié a donde pertenece, reemplazándolo con HttpClient de Java.

Para las pruebas, tomé el servicio https://jsonplaceholder.typicode.com/ , que me sugirió un desarrollador conocido. Guardé un enlace que emite Json con comentarios a una variable, para no duplicar e iniciar pruebas.

const val URL = "https://jsonplaceholder.typicode.com/comments"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

La función estaba lista e incluso funcionando. Llegaron los datos.

fun getDataAsync(url: String): String? {
    val httpClient = HttpClient.newBuilder()
        .build()
    val httpRequest = HttpRequest.newBuilder()
        .uri(URI.create(link)).build()

    return httpClient.sendAsync(httpRequest, BodyHandlers.ofString())
        .join().body()
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Ahora era necesario comprobar la velocidad de trabajo. Armado con MeasureTimeMillis, ejecuté el código.

val asyncTime = measureTimeMillis { 
    val res = (1..10)
        .toList()
        .map {getDataAsync("$URL/$it")}
    res.forEach { println(it) }
}
println("   $asyncTime ")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Todo funcionó como debería, pero lo quería más rápido. Después de investigar un poco en Internet, encontré una solución en la que las tareas se realizan en paralelo.

Mapa paralelo

, . , , .

suspend fun <A, B> Iterable<A>.pmap(f: suspend (A) -> B): List<B> =
    coroutineScope {
        map { async { f(it) } }.awaitAll()
    }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, ( Iterable) pmap, . A. async , .awaitAll() . suspend, .

, , - .

val parmapTime = measureTimeMillis {
    runBlocking {
        val res = (1..10)
            .toList()
            .pmap { getDataAsync("$URL/$it") }
        println(mapResult)
    }
}
println(" pmap $parmapTime ")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

- 1523, . map async, .

Parallel Map v 2.0

, , .

suspend fun <T, V> Iterable<T>.parMap(func: suspend (T) -> V): Iterable<V> =
    coroutineScope {
        map { element -> 
            async(Dispatchers.IO) { func(element) } 
        }.awaitAll() 
    }

val parMapTime = measureTimeMillis {
    runBlocking {
        val res = (1..10)
            .toList()
            .parMap { getDataAsync("$URL/$it") }
    }
    println(res)
}
println(" map  $parMapTime ")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Dispatchers.IO 2 ~ 610 . ! ( , excel ..) . , - .

Java ParallelStream

, stackowerflow parallelStream. , IDEA.

val javaParallelTime = measureTimeMillis { 
    val res = (1..10).toList()
        .parallelStream()
        .map { getDataAsync("$URL/$it") }
    res.forEach { println(it) }
}
println("Java parallelSrtream  $javaParallelTime ")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . , . stream . , , , "" , Json.

, - , async . .

Los resultados se pueden ver en la siguiente tabla. Para mí, definitivamente decidí dejar async en espera . Principalmente por el manejo de errores más simple, por supuesto. Y no hay necesidad de ir más allá de las corrutinas aquí.

En el futuro, hay pensamientos para organizar esto en una pequeña biblioteca y usarlo para propósitos personales y, por supuesto, reescribirlo todo desde el "código hindú" al humano, siempre que haya suficientes posibilidades. Y luego cárguelo todo en vds.

Espero que mi experiencia sea de utilidad para alguien. ¡Me encantaría recibir críticas y consejos constructivos! Gracias a todos