Ограничение максимального количества сопрограмм, которые могут выполняться в области

Question

Я перевожу наше текущее приложение с Java на Kotlin, и я столкнулся с этой проблемой.

Реализация Java, используемая для использования потоков для передачи данных с сервера.

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

При переводе в Kotlin я использовал Coroutines

Это создает проблему, потому что, по-видимому, сервер не может обработать фактически отправленные 100 запросов.

Все сопрограммы запускаются в одной и той же области, поэтому это выглядит примерно так:

//this is a custom scope that launches on Dispatchers.IO + a job that I can use to cancel everything
transferScope.launch {
     //loadData is a suspending function that returns true/false 
     val futures = mDownloadJobs.map{ async { it.loadData() } }
     val responses = futures.awaitAll()
     //check that everything in responses is true etc....
}

Есть ли способ заставить конкретный TransferScope разрешать одновременно только до 5 сопрограмм, а затем, когда один заканчивает, запускать другой? (Меня не волнует порядок)

Если это невозможно сделать с помощью объема, есть ли другой способ добиться этого?

Answer 1

Требуется, чтобы каждая сопрограмма получила разрешение Kotlin Semaphore из 5 разрешений, прежде чем делать запрос.

Примерно так:

    import kotlinx.coroutines.sync.Semaphore

    val requestSemaphore = Semaphore(5)

    val futures = mDownloadJobs.map {
        async {
            // Will limit number of concurrent requests to 5
            requestSemaphore.withPermit {
                it.loadData()
            }
        }
    }

    val responses = futures.awaitAll()

Answer 2

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

requests.chunked(4).forEachIndexed { index, chunk ->
    coroutineScope {
        LOG("processing chunk $index")
        chunk.forEach {
            launch {
                delay(100)
            }
        }
        LOG("done processing $index")
    }
}