Добрый день! Подскажите пожалуйста по корутинам, при использовании JMS разбираю сообщения асинхронно, код имеет примерно следующий вид:

ovveride fun onMessage(message: Message) = runBlocking {
   worker()
}

fun worker(msg: String) = GlobalScope.launch
{
    doSomeWork(msg)//какой-то долгий процесс  
}

Как лучше написать, чтобы была возможность контролировать количество одновременно работающих worker?

runBlocking { launch { ... } }

заблокирует текущий поток до конца выполнения корутины, потому такая конструкция не имеет смысла.

GlobalScope ведь, не заблокирует. Проблема в другом, для примера скажем, что сообщения весят по 100mb и единовременно их может быть от 1 до 10000 и над ними выполняются тяжелые операции. Хочу контролировать количество единовременно обрабатываемых сообщений, явно указать что число одновременных обработчиков 100 и блокировать onMessage если 100 в работе, и брать следующее сообщение только когда один из worker'ов закончит работу.

GlobalScope

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

runBlocking

. Если ограничить нужно до разумного количества, и есть возможность выделить количество потоков, соответствующее максимальному количеству параллельных задач, то это можно сделать на основании

newFixedThreadPoolContext

-- достаточно запустить на нём

actor

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

AtomicInt

и врапая свой

doSomeWork

в инкремент и декремент счётчика.

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

Да, я выпил кофе и таки сообразил, что

runBlocking

в данном случае ничего не делает, потому что

GlobalScope.launch

сам не саспендится, а это единственная операция в

onMessage

. То есть

runBlocking

можно вообще со спокойной душой убрать. Правда, вы всё ещё, оперируя глобальным скоупом, теряете исключения и возможность сделать graceful shutdown, так как у вас нету конкретного скоупа, в котором запущены все задачи. Можете конкретную ссылку подкинуть, о чём речь идёт? Там несколько часовых докладов всё-таки.

Если про доклад, то я об этом

А какая цель? Чего хочется достичь?

@elizarov Роман, добрый день! Идея такая, есть MQ очередь из нее прилетают большие текстовые сообщения в заархивированном виде, далее они обрабатываются на бэке - распаковываются, парсятся из них извлекается нужная инфа, происходят обращения по JDBC, по REST итд. После того как сообщение обработано необходимо отправить ответ об успешной или неуспешной обработке. И мне кажется Ваш кейс отлично укладывается в эту концепцию, я хочу обрабатывать сообщения асинхронно и при этом контролировать количество одновременно обрабатываемых сообщений, после обработки отправлять ответ. Вот ссылка это Ваш пример с презентации, только я его переделал для работы со строками. И он у меня не работает, когда количество worker’ов меньше количества сообщений. https://try.kotlinlang.org/#/UserProjects/qk78mcnfm1m16fnf5m6jqrj9ts/djs2m87ma852b3qa6pou966565

Вижу одну потенциальную проблему — если все workers заняты (никто не пытается взять задание из locations), то downloader будет ждать на locations.send(…), не сможет сделать contents.recieve, а значит и worker не сможет освободиться (не сможет сделать contents.receiver) — вот вам и дед лок

Роман, спасибо за ответ! Все получилось. Я правильно понимаю, что в какой-то момент даже при таком подходе может возникнуть дедлок? Хотел попробовать в production пустить этот кейс (знаю, что experimental), в любом случае будет много тестов на тестовых средах, если все будет ок - может и пойдет в прод.

Если сделать как я сказал (и не забыть первым поставить

references.onReceive

), то легко доказать, что дедлока быть не может.

Ясно, буду экспериментировать) :kotlin:👍