Канал Андрея про бекенд

Недавно у меня возникла следующая задача: необходимо запросить данные у трех сервисов, но достаточно получить первый валидный ответ от любого их них, остальные 2 вызова можно не ожидать.

🤝 Интерфейс вызова следующий:

fun performCall(key: Key) : CompletableFuture<Value>

🤯 Я попытылся написать это на java, но ничего адекватного по количеству строк и общей читаемости кода у меня не получилось.

🐣 Решение я нашел в использовании корутин. Первое, что я сделал: превратил CompletableFuture<Value> в Deferred<Value>, используя extension-функцию public fun <T> CompletionStage<T>.asDeferred(): Deferred<T>. Это необходимо, чтобы переместить наш код в плоскость корутин.

🪢 Далее я использовал интересную функцию - select. Ей на вход можно передать любое количество “ожидающих (suspended)” функций и select будет ждать, пока любая из этих функций не вернет результат. Тогда исполнение продолжится и select вернет вам результат этой "победившей" корутины.

🍎 Таким образом, моя задача сводится к довольно простой вещи - поместим все наши вызовы в список и передадим функции select, а она вернет нам первый полученный результат. Это код, который нам нужен:

val calls = services.map { performCall(key).asDeferred() }.toList()
val res = select {
    calls.forEach { call ->
        call.onAwait { it } // it это результат вызова
    }
}

💩 В реальности сервисы могут кидать исключения или возвращать пустой результат null. А нам необходимо дождаться именно первого валидного результата. Мы можем сделать следующее - проанализируем результат и в случае null будем снова заходить в select.

🚨 И тут надо помнить еще один интересный нюанс: если вы снова передадите в select тот же самый список, то он повторно выдаст вам результат ошибочного вызова, ведь он и правда завершен :) Поэтому давайте удалять из списка те вызовы, которые выполнились и не являются валидными. Только не забывайте поместить их в thread-safe контейнер, иначе легко словите ConcurrentModificationException.

Финальное решение выглядит как-то так (за исключением try-catch, чтобы сократить):

val calls = services.map { performCall(key).asDeferred() }.toCollection(ConcurrentHashMap.newKeySet())

var res: Value? = null
while (calls.isNotEmpty()) {
    val res = select {
        calls.forEach { call ->
            call.onAwait { callRes ->
                calls.remove(call)
                callRes 
            } 
        }
    }
    if (res != null) break
}

☄️ Отмечу еще, что select бывает невероятно полезен в сочетании с использованием каналов (kotlinx.coroutines.Channel). Канал - аналог BlockingQueue в мире корутин, позже выложу пост о них. Если у вас есть задача мониторить множество очередей на предмет новых элементов и сразу же передавать их на обработку, то лучше средства не найти.

🪐 Интересно, что функция select по свой сути является аналогом системного вызова select на основе которого реализован "неблокирующий" ввод-вывод. Он позволяет мониторить множество TCP соединений на предмет новых событий IO и превращать их в единый "уплотненный" или иначе "мультиплексированный" поток событий, что позволяет очень эффективно использовать ресурсы системы. О "мультиплексировании" уже говорили тут и будем говорить еще.

Ставьте 🔥, если было интересно. Если вы знаете код на java, который решает аналогичную задачу и является плюс-минус таким же лаконичным, делитесь в комментах!

#kotlin #coroutines #java

Please open Telegram to view this post

VIEW IN TELEGRAM

Please open Telegram to view this post

VIEW IN TELEGRAM

🔥37👍7❤3🆒1

3.05K views09:35