10.02.2021 14:19
foxcode85 14 1500 Источник

Всем привет! В своей работе я часто использую Kotlin для автоматизации. Деятельность моя не связана напрямую с программированием, но Котлин здорово упрощает рабочие задачи.

Недавно нужно было собрать данные немаленького размера, дабы сделать анализ, поэтому решил написать небольшой скрипт, для получения данных и сохранения их в Excel. С последним пунктом проблем не возникло - почитал про Apache POI, взял пару примеров из официальной документации, доработав под себя. Чего не скажешь про запросы в Сеть.

Источник отдавал пачками json и надо было как-то быстро эти "пачки" собирать, преобразовывая в текст и записывая в файл таблицу.

Асинхронный метод

Начать решил с простой асинхронщины. Немного поковыряв HttpUrlConnection, отправил туда, где ему и место, заменив на HttpClient из Java.

Для тестов взял сервис https://jsonplaceholder.typicode.com/, который мне подсказал один знакомый разработчик. Сохранил ссылку, которая выдает Json с комментариями в переменную, дабы не дублировать и начал тесты.

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

Функция была готова и даже работала. Данные приходили.

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()
}

Теперь надо было проверить скорость работы. Вооружившись measureTimeMillis я запустил код. 

val asyncTime = measureTimeMillis { 
    val res = (1..10)
        .toList()
        .map {getDataAsync("$URL/$it")}
    res.forEach { println(it) }
}
println("Асинхронный запрос время $asyncTime мс")

Все работало как надо, но хотелось быстрее. Немного покопавшись в Интернете, я набрел на решение, в котором задачи выполняются параллельно.

Parallel Map

Автор в своем блоге пишет, что код ниже выполняется параллельно с использованием корутин. Ну что, я давно хотел их попробовать, а тут представилась возможность. 

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 таких проблем не возникло. Там можно контролировать данные на каждом шаге итерации и удобно обрабатывать ошибки.

Выводы

Результаты можно посмотреть в таблице ниже. Для себя я однозначно решил оставить async await. В основном конечно из-за более простой обработки ошибок. Да и за пределы корутин тут выходить не надо.

Метод

Время (ms)

Асинхронный метод

1487

Реализация pmap из Сети

1523

Мой вариант - parallelMap

610

Java.parallelStream

578

getMyData (with @zeldigas)

~ 498

В дальнейшем, есть мысли оформить это в небольшую библиотеку и использовать в личных целях, и конечно переписать все это с "индусского кода" на человеческий, на сколько хватит возможностей. А потом залить все это на vds.

Надеюсь мой опыт кому-нибудь пригодится. Буду рад конструктивной критике и советам! Всем спасибо

P.S.

Спасибо товарищу @zeldigas за подсказку насчет HttpClient! Как оказалось, вызов join() блокирует выполнение. Он предложил подключить kotlinx-coroutines-jdk8 с экстеншенами для CompletableFuture, но я сделал через suspendCoroutine и результат получился аналогичным.

suspend fun getMyData(httpClient: HttpClient, url: String): String? = 
    suspendCoroutine {
        val httpRequest = HttpRequest.newBuilder().uri(URI.create(url)).build()
        httpClient.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofString())
            .thenApply { obj -> it.resume(obj.body()) }
}

Наша suspendCoroutine приостанавливает выполнение корутины, дожидается результата и принимает его в методе resume.

val httpClient:HttpClient = HttpClient.newBuilder().build()
val result = measureTimeMillis {
    runBlocking {
        val res =(1..10).toList()
            .map { async { getMyData(httpClient, "$URL/$it") } }
            .awaitAll()
        println(res)
    }
}
println("Time for requests: $result")

В итоге, результат получился в районе 498 мс, что быстрее чем остальные варианты. Этот случай еще раз доказывает, что комментарии на Хабре порой очень полезны.

Всем спасибо! Всем пока!