Подпрограммы и каналы — учебное пособие

Перед началом

1. Загрузите и установите последнюю версию IntelliJ IDEA.

2. Скопируйте шаблон проекта шаблон проекта, выбрав Получить из VCS на экране «Добро пожаловать» или выбрав Файл | Новый | Проект из системы контроля версий.

   Вы также можете скопировать его из командной строки:

   git clone https://github.com/kotlin-hands-on/intro-coroutines
   

Блокирующие запросы

Для выполнения HTTP-запросов к GitHub вы будете использовать библиотеку Retrofit. Она позволяет запросить список репозиториев в рамках данной организации и список авторов для каждого репозитория:

interface GitHubService {
    @GET("orgs/{org}/repos?per_page=100")
    fun getOrgReposCall(
        @Path("org") org: String
    ): Call<List<Repo>>

    @GET("repos/{owner}/{repo}/contributors?per_page=100")
    fun getRepoContributorsCall(
        @Path("owner") owner: String,
        @Path("repo") repo: String
    ): Call<List<User>>
}

Этот API используется функцией loadContributorsBlocking() для получения списка авторов для данной организации.

1. Откройте src/tasks/Request1Blocking.kt для просмотра его реализации:

   fun loadContributorsBlocking(service: GitHubService, req: RequestData): List<User> {
       val repos = service
           .getOrgReposCall(req.org)   // #1
           .execute()                  // #2
           .also { logRepos(req, it) } // #3
           .body() ?: emptyList()      // #4
   
       return repos.flatMap { repo ->
           service
               .getRepoContributorsCall(req.org, repo.name) // #1
               .execute()                                   // #2
               .also { logUsers(repo, it) }                 // #3
               .bodyList()                                  // #4
       }.aggregate()
   }
   

   • Сначала вы получаете список репозиториев для данной организации и сохраняете его в списке repos. Затем для каждого репозитория запрашивается список авторов, и все списки объединяются в один итоговый список авторов.

   • getOrgReposCall() и getRepoContributorsCall() оба возвращают экземпляр класса *Call (#1). На этом этапе ни один запрос не отправляется.

   • Затем вызывается *Call.execute() для выполнения запроса (#2). execute() — это синхронный вызов, который блокирует подлежащую нить.

   • При получении ответа результат регистрируется, вызывая функции logRepos() и logUsers() (#3). Если HTTP-ответ содержит ошибку, эта ошибка будет здесь зарегистрирована.

   • Наконец, получите тело ответа, содержащее необходимые данные. Для этого учебника вы будете использовать пустой список в качестве результата в случае ошибки и регистрировать соответствующую ошибку (#4).

2. Чтобы избежать повторения .body() ?: emptyList(), объявляется функция-расширение bodyList().

   fun <T> Response<List<T>>.bodyList(): List<T> {
       return body() ?: emptyList()
   }
   

3. Запустите программу снова и посмотрите вывод системы в IntelliJ IDEA. Он должен выглядеть примерно так:

   1770 [AWT-EventQueue-0] INFO  Contributors - kotlin: loaded 40 repos
   2025 [AWT-EventQueue-0] INFO  Contributors - kotlin-examples: loaded 23 contributors
   2229 [AWT-EventQueue-0] INFO  Contributors - kotlin-koans: loaded 45 contributors
   ...
   

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

   • Последним элементом в каждой строке является фактическое сообщение: сколько репозиториев или авторов было загружено.

   Этот вывод логов демонстрирует, что все результаты были зарегистрированы из основного потока. При запуске кода с параметром БЛОКИРУЮЩИЙ окно замирает и не реагирует на входные данные до завершения загрузки. Все запросы выполняются из того же потока, что и вызов loadContributorsBlocking(), который является основным потоком пользовательского интерфейса (в Swing это поток диспетчеризации событий AWT). Этот основной поток блокируется, и поэтому пользовательский интерфейс замирает:

   

   После загрузки списка авторов результат обновляется.

4. В src/contributors/Contributors.kt, найдите функцию loadContributors(), отвечающую за выбор способа загрузки авторов, и посмотрите, как вызывается loadContributorsBlocking().

   when (getSelectedVariant()) {
       BLOCKING -> { // Blocking UI thread
           val users = loadContributorsBlocking(service, req)
           updateResults(users, startTime)
       }
   }
   

   • Вызов updateResults() идет сразу после вызова loadContributorsBlocking().

   • updateResults() обновляет пользовательский интерфейс, поэтому его всегда следует вызывать из потока пользовательского интерфейса.

   • Поскольку loadContributorsBlocking() также вызывается из потока пользовательского интерфейса, поток пользовательского интерфейса блокируется, и пользовательский интерфейс замирает.

Задача 1

Первая задача поможет вам ознакомиться с областью задачи. В настоящее время имя каждого автора повторяется несколько раз, один раз для каждого проекта, в котором они участвовали. Реализуйте функцию aggregate() для объединения пользователей таким образом, чтобы каждый автор добавлялся только один раз. Свойство User.contributions должно содержать общее количество вкладов данного пользователя во все проекты. Результирующий список должен быть отсортирован по убыванию количества вкладов.

Откройте src/tasks/Aggregation.kt и реализуйте функцию List<User>.aggregate(). Пользователи должны быть отсортированы по общему количеству их вкладов.

Соответствующий тестовый файл test/tasks/AggregationKtTest.kt показывает пример ожидаемого результата.

После выполнения этой задачи результирующий список для организации «kotlin» должен быть похож на следующий:

Решение задачи 1

1. Чтобы сгруппировать пользователей по логину, используйте groupBy(), которая возвращает карту из логина к всем появлениям пользователя с этим логином в разных репозиториях.

2. Для каждого элемента карты подсчитайте общее количество вкладов для каждого пользователя и создайте новый экземпляр класса User с заданным именем и общим количеством вкладов.

3. Отсортируйте результирующий список по убыванию:

   fun List<User>.aggregate(): List<User> =
       groupBy { it.login }
           .map { (login, group) -> User(login, group.sumOf { it.contributions }) }
           .sortedByDescending { it.contributions }
   

Альтернативным решением является использование функции groupingBy() вместо groupBy().

Обработка событий (Callbacks)

Предыдущее решение работает, но блокирует поток, из-за чего интерфейс пользователя зависает. Традиционный подход, который этого избегает, использует обработку событий (callbacks).

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

Для обеспечения отзывчивости интерфейса пользователя, можно либо перенести всю вычисление в отдельный поток, либо переключиться на API Retrofit, который использует обработку событий (callbacks) вместо блокирующих вызовов.

thread {
    updateResults(loadContributorsBlocking(service, req))
}

Использование API Retrofit с обработкой событий

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

Обработка данных для каждого репозитория должна быть разделена на две части: загрузка и обработка полученного ответа. Вторая часть обработки должна быть выделена в callback.

Загрузку для каждого репозитория можно начать до получения результата для предыдущего репозитория (и вызова соответствующего callback):

API Retrofit с обработкой событий может помочь в этом. Функция Call.enqueue() запускает HTTP-запрос и принимает callback в качестве аргумента. В этом callback необходимо указать, что нужно сделать после каждого запроса.

Откройте src/tasks/Request3Callbacks.kt и изучите реализацию loadContributorsCallbacks(), которая использует этот API:

fun loadContributorsCallbacks(
    service: GitHubService, req: RequestData,
    updateResults: (List<User>) -> Unit
) {
    service.getOrgReposCall(req.org).onResponse { responseRepos ->  // #1
        logRepos(req, responseRepos)
        val repos = responseRepos.bodyList()

        val allUsers = mutableListOf<User>()
        for (repo in repos) {
            service.getRepoContributorsCall(req.org, repo.name)
                .onResponse { responseUsers ->  // #2
                    logUsers(repo, responseUsers)
                    val users = responseUsers.bodyList()
                    allUsers += users
                }
            }
        }
        // TODO: Why doesn't this code work? How to fix that?
        updateResults(allUsers.aggregate())
    }

• Для удобства этот фрагмент кода использует функцию расширения onResponse(), объявленную в том же файле. Она принимает лямбда-выражение в качестве аргумента вместо объекта выражения.

• Логика обработки ответов выделена в callbacks: соответствующие лямбда-выражения начинаются на строках #1 и #2.

Однако, приведенное решение не работает. Если вы запустите программу и загрузите участников, выбрав опцию CALLBACKS, ничего не отобразится. Однако тесты, которые сразу возвращают результат, проходят.

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

val allUsers = mutableListOf<User>()
for ((index, repo) in repos.withIndex()) {   // #1
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            logUsers(repo, responseUsers)
            val users = responseUsers.bodyList()
            allUsers += users
            if (index == repos.lastIndex) {    // #2
                updateResults(allUsers.aggregate())
            }
        }
}

val allUsers = Collections.synchronizedList(mutableListOf<User>())
val numberOfProcessed = AtomicInteger()
for (repo in repos) {
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            logUsers(repo, responseUsers)
            val users = responseUsers.bodyList()
            allUsers += users
            if (numberOfProcessed.incrementAndGet() == repos.size) {
                updateResults(allUsers.aggregate())
            }
        }
}

val countDownLatch = CountDownLatch(repos.size)
for (repo in repos) {
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            // processing repository
            countDownLatch.countDown()
        }
}
countDownLatch.await()
updateResults(allUsers.aggregate())

Функции приостановки

Вы можете реализовать ту же логику, используя функции приостановки. Вместо возвращения Call<List<Repo>>, определите вызов API как функцию приостановки следующим образом:

interface GitHubService {
    @GET("orgs/{org}/repos?per_page=100")
    suspend fun getOrgRepos(
        @Path("org") org: String
    ): List<Repo>
}

• getOrgRepos() определяется как функция suspend. Когда вы используете функцию приостановки для выполнения запроса, подлежащая нить не блокируется. Более подробные сведения о том, как это работает, будут представлены в последующих разделах.

• getOrgRepos() возвращает результат напрямую, вместо возвращения Call. Если результат неудачный, выбрасывается исключение.

В качестве альтернативы, Retrofit позволяет возвращать результат, обернутый в Response. В этом случае тело результата предоставляется, и можно вручную проверить наличие ошибок. Этот учебник использует версии, которые возвращают Response.

В src/contributors/GitHubService.kt, добавьте следующие объявления в интерфейс GitHubService:

interface GitHubService {
    // getOrgReposCall & getRepoContributorsCall declarations

    @GET("orgs/{org}/repos?per_page=100")
    suspend fun getOrgRepos(
        @Path("org") org: String
    ): Response<List<Repo>>

    @GET("repos/{owner}/{repo}/contributors?per_page=100")
    suspend fun getRepoContributors(
        @Path("owner") owner: String,
        @Path("repo") repo: String
    ): Response<List<User>>
}

suspend fun loadContributorsSuspend(service: GitHubService, req: RequestData): List<User> {
    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    return repos.flatMap { repo ->
        service.getRepoContributors(req.org, repo.name)
            .also { logUsers(repo, it) }
            .bodyList()
    }.aggregate()
}

Сопрограммы

Код с функциями приостановки выглядит схоже с "блокирующей" версией. Основное отличие от блокирующей версии заключается в том, что вместо блокирования нити сопрограмма приостанавливается:

block -> suspend
thread -> coroutine

Запуск новой сопрограммы

Если вы посмотрите, как loadContributorsSuspend() используется в src/contributors/Contributors.kt, вы увидите, что она вызывается внутри launch . launch — это функция библиотеки, которая принимает лямбда-выражение в качестве аргумента:

launch {
    val users = loadContributorsSuspend(req)
    updateResults(users, startTime)
}

Здесь launch запускает новую вычислительную задачу, ответственную за загрузку данных и отображение результатов. Вычисление является приостанавливаемым — при выполнении сетевых запросов оно приостанавливается и освобождает подлежащую нить. При возвращении результата сетевого запроса вычисление возобновляется.

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

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

Когда вычисление готово к продолжению, оно возвращается в нить (не обязательно в ту же самую).

В примере loadContributorsSuspend() каждый запрос "contributors" теперь ожидает результата, используя механизм приостановки. Сначала отправляется новый запрос. Затем, в ожидании ответа, вся сопрограмма "загрузка участников", запущенная функцией launch , приостанавливается.

Сопрограмма возобновляется только после получения соответствующего ответа:

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

1. Запустите программу, используя опцию ПРИОСТАНОВИТЬ. Лог подтверждает, что все запросы отправляются в основную нить пользовательского интерфейса:

   2538 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 30 repos
   2729 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - ts2kt: loaded 11 contributors
   3029 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin-koans: loaded 45 contributors
   ...
   11252 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin-coroutines-workshop: loaded 1 contributors
   

2. Лог может показать, в какой сопрограмме выполняется соответствующий код. Для этого откройте Запуск | Редактировать конфигурации и добавьте параметр VM -Dkotlinx.coroutines.debug:

   

   Имя сопрограммы будет добавлено к имени нити при запуске main() с этим параметром. Вы также можете изменить шаблон запуска всех файлов Kotlin и включить этот параметр по умолчанию.

Теперь весь код выполняется в одной сопрограмме, "загрузке участников" упомянутой выше, обозначенной как @coroutine#1 . Ожидая результата, не следует повторно использовать нить для отправки других запросов, так как код написан последовательно. Новый запрос отправляется только после получения предыдущего результата.

Функции приостановки обращаются с нитью справедливо и не блокируют ее в ожидании. Однако это пока не вносит никакой параллельности.

Асинхронность

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

Чтобы запустить новую корутину, используйте один из основных строителей корутин: launch, async, или runBlocking. Различные библиотеки могут определять дополнительные строители корутин.

async запускает новую корутину и возвращает объект Deferred. Deferred представляет собой концепцию, известную и другими названиями, такими как Future или Promise. Она хранит вычисление, но откладывает момент получения конечного результата; она обещает результат в некотором будущем.

Основное отличие между async и launch заключается в том, что launch используется для запуска вычисления, от которого не ожидается получения конкретного результата. launch возвращает Job, представляющую корутину. Можно дождаться её завершения, вызвав Job.join().

Deferred — это универсальный тип, который расширяет Job. Вызов async может вернуть Deferred<Int> или Deferred<CustomType>, в зависимости от того, что возвращает лямбда-выражение (последнее выражение внутри лямбды — результат).

Чтобы получить результат корутины, можно вызвать await() на экземпляре Deferred. Пока ожидается результат, корутина, из которой вызван этот await(), приостанавливается:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferred: Deferred<Int> = async {
        loadData()
    }
    println("waiting...")
    println(deferred.await())
}

suspend fun loadData(): Int {
    println("loading...")
    delay(1000L)
    println("loaded!")
    return 42
}

runBlocking используется как мост между обычными и приостанавливаемыми функциями, или между блокирующим и неблокирующим мирами. Он работает как адаптер для запуска главной корутины верхнего уровня. Он предназначен в основном для использования в функциях и тестах main().

Если есть список объектов отложенного выполнения, можно вызвать awaitAll() для ожидания результатов всех из них:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferreds: List<Deferred<Int>> = (1..3).map {
        async {
            delay(1000L * it)
            println("Loading $it")
            it
        }
    }
    val sum = deferreds.awaitAll().sum()
    println("$sum")
}

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

Общее время загрузки примерно такое же, как в версии CALLBACKS, но она не требует никаких обратных вызовов. Более того, async чётко выделяет части кода, выполняющиеся параллельно.

suspend fun loadContributorsConcurrent(
    service: GitHubService,
    req: RequestData
): List<User> = coroutineScope {
    // ...
}

val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
    async {
        // load contributors for each repo
    }
}
deferreds.awaitAll() // List<List<User>>

Структурированная конкуратность

• Область корутины отвечает за структуру и отношения «родитель-потомок» между различными корутинами. Новые корутины обычно необходимо запускать внутри области.

• Контекст корутины хранит дополнительную техническую информацию, используемую для запуска данной корутины, например, пользовательское имя корутины или диспетчер, определяющий потоки, на которых корутина должна планироваться.

При использовании launch, async, или runBlocking для запуска новой корутины, они автоматически создают соответствующую область. Все эти функции принимают лямбда-выражение с получателем в качестве аргумента, и CoroutineScope является неявным типом получателя:

launch { /* this: CoroutineScope */ }

• Новые корутины могут запускаться только внутри области.

• launch и async объявляются как расширения для CoroutineScope, поэтому неявный или явный получатель всегда должен передаваться при их вызове.

• Корутина, запущенная с помощью runBlocking, является единственным исключением, потому что runBlocking определена как функция верхнего уровня. Но поскольку она блокирует текущий поток, она предназначена в основном для использования в функциях main() и тестах в качестве мостовой функции.

Новая корутина внутри runBlocking, launch, или async запускается автоматически внутри области:

import kotlinx.coroutines.*

fun main() = runBlocking { /* this: CoroutineScope */
    launch { /* ... */ }
    // the same as:   
    this.launch { /* ... */ }
}

При вызове launch внутри runBlocking, она вызывается как расширение для неявного получателя типа CoroutineScope. В качестве альтернативы, вы можете явно написать this.launch.

Вложенная корутина (запущенная с помощью launch в этом примере) может рассматриваться как дочерняя корутина внешней корутины (запущенной с помощью runBlocking). Это отношение «родитель-потомок» реализуется через области; дочерняя корутина запускается из области, соответствующей родительской корутине.

Можно создать новую область без запуска новой корутины, используя функцию coroutineScope. Чтобы структурированно запускать новые корутины внутри функции suspend без доступа к внешней области, вы можете создать новую область корутины, которая автоматически становится дочерней областью внешней области, из которой вызывается функция suspend. loadContributorsConcurrent() - хороший пример.

Вы также можете запустить новую корутину из глобальной области, используя GlobalScope.async или GlobalScope.launch. Это создаст независимую корутину верхнего уровня.

Механизм структуры корутин называется структурированной конкуратностью. Она предоставляет следующие преимущества по сравнению с глобальными областями:

• Область, как правило, отвечает за дочерние корутины, срок жизни которых привязан к сроку жизни области.

• Область может автоматически отменять дочерние корутины, если что-то идет не так или пользователь меняет решение и решает отменить операцию.

• Область автоматически ожидает завершения всех дочерних корутин. Поэтому, если область соответствует корутине, родительская корутина не завершается до тех пор, пока все корутины, запущенные в ее области, не завершатся.

При использовании GlobalScope.async, нет структуры, которая связывает несколько корутин в более узкой области. Корутины, запущенные из глобальной области, являются независимыми - их срок жизни ограничен только сроком жизни всего приложения. Можно сохранить ссылку на запущенную из глобальной области корутину и дождаться ее завершения или явно отменить ее, но это не произойдет автоматически, как в случае со структурированной конкуратностью.

val job = launch { }
job.setUpCancellation()

Использование контекста внешней области

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

Теперь пришло время узнать, как работает использование диспетчера из внешней области. Новая область, созданная функцией coroutineScope или конструкторами корутин, всегда наследует контекст из внешней области. В этом случае внешняя область - это область, из которой была вызвана функция suspend loadContributorsConcurrent():

launch(Dispatchers.Default) {  // outer scope
    val users = loadContributorsConcurrent(service, req)
    // ...
}

Все вложенные корутины автоматически запускаются с унаследованным контекстом. Диспетчер является частью этого контекста. Поэтому все корутины, запущенные функцией async, запускаются с контекстом по умолчанию диспетчера:

suspend fun loadContributorsConcurrent(
    service: GitHubService, req: RequestData
): List<User> = coroutineScope {
    // this scope inherits the context from the outer scope
    // ...
    async {   // nested coroutine started with the inherited context
        // ...
    }
    // ...
}

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

Отображение прогресса

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

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

Для реализации этой функциональности в src/tasks/Request6Progress.kt, вам необходимо передать логику обновления интерфейса в качестве обратного вызова, чтобы он вызывался на каждом промежуточном шаге:

suspend fun loadContributorsProgress(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
    // loading the data
    // calling `updateResults()` on intermediate states
}

В месте вызова в Contributors.kt, обратный вызов передается для обновления результатов из потока Main для опции PROGRESS:

launch(Dispatchers.Default) {
    loadContributorsProgress(service, req) { users, completed ->
        withContext(Dispatchers.Main) {
            updateResults(users, startTime, completed)
        }
    }
}

• Параметр updateResults() объявлен как suspend в loadContributorsProgress(). Необходимо вызвать withContext, которая является функцией suspend внутри соответствующего лямбда-аргумента.

• Обратный вызов updateResults() принимает дополнительный булевый параметр в качестве аргумента, указывающий, завершена ли загрузка и являются ли результаты окончательными.

Задача 6

В файле Request6Progress.kt реализуйте функцию loadContributorsProgress(), которая показывает промежуточный прогресс. Основывайтесь на функции loadContributorsSuspend() из Request4Suspend.kt.

• Используйте простую версию без конкурентности; вы добавите её позже в следующем разделе.

• Промежуточный список участников должен отображаться в «агрегированном» состоянии, а не только список пользователей, загруженных для каждого репозитория.

• Общее количество вкладов для каждого пользователя должно увеличиваться при загрузке данных для каждого нового репозитория.

Решение задачи 6

Для хранения промежуточного списка загруженных участников в «агрегированном» состоянии определите переменную allUsers, которая хранит список пользователей, а затем обновите её после загрузки участников для каждого нового репозитория:

suspend fun loadContributorsProgress(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    var allUsers = emptyList<User>()
    for ((index, repo) in repos.withIndex()) {
        val users = service.getRepoContributors(req.org, repo.name)
            .also { logUsers(repo, it) }
            .bodyList()

        allUsers = (allUsers + users).aggregate()
        updateResults(allUsers, index == repos.lastIndex)
    }
}

Последовательное против параллельного

Обратный вызов updateResults() вызывается после завершения каждого запроса:

Этот код не включает конкурентности. Он последовательный, поэтому вам не нужна синхронизация.

Лучшим вариантом было бы отправить запросы параллельно и обновить промежуточные результаты после получения ответа для каждого репозитория:

Для добавления конкурентности используйте каналы.

Каналы

Написание кода с общей изменяемой областью состояния довольно сложно и подвержено ошибкам (как в решении с обратными вызовами). Более простой способ — передавать информацию посредством связи, а не с помощью общей изменяемой области. Корутины могут обмениваться данными через каналы.

Каналы — это примитивы связи, которые позволяют передавать данные между корутинами. Одна корутина может отправить некоторую информацию в канал, а другая может получить эту информацию из него:

Корутина, которая отправляет (производит) информацию, часто называется производителем, а корутина, которая получает (потребляет) информацию, называется потребителем. Одна или несколько корутин могут отправить информацию в один и тот же канал, и одна или несколько корутин могут получить данные из него:

Когда много корутин получают информацию из одного и того же канала, каждый элемент обрабатывается только один раз одной из корутин-потребителей. После обработки элемент немедленно удаляется из канала.

Вы можете представить канал как нечто подобное набору элементов или, точнее, очереди, в которой элементы добавляются с одного конца и извлекаются с другого. Однако есть важное различие: в отличие от коллекций, даже в их синхронизированных версиях, канал может подвешиваться send() и receive() операциями. Это происходит, когда канал пуст или полон. Канал может быть полным, если размер канала имеет верхнюю границу.

Channel представлен тремя различными интерфейсами: SendChannel, ReceiveChannel, и Channel, причём последний расширяет первые два. Обычно вы создаёте канал и передаёте его производителям как SendChannel экземпляр, чтобы только они могли отправлять информацию в канал. Вы передаёте канал потребителям как ReceiveChannel экземпляр, чтобы только они могли получать из него. Оба send и receive метода объявлены как suspend.

interface SendChannel<in E> {
    suspend fun send(element: E)
    fun close(): Boolean
}

interface ReceiveChannel<out E> {
    suspend fun receive(): E
}

interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

Производитель может закрыть канал, чтобы указать, что больше элементов не поступает.

В библиотеке определены несколько типов каналов. Они различаются тем, сколько элементов они могут хранить внутри и может ли вызов send() быть приостановлен или нет. Для всех типов каналов вызов receive() ведёт себя аналогично: он получает элемент, если канал не пуст; в противном случае он приостанавливается.

Неограниченный канал

Неограниченный канал — это ближайший аналог очереди: производители могут отправлять элементы в этот канал, и он будет неограниченно расти. Вызов send() никогда не будет приостановлен. Если программе не хватит памяти, вы получите OutOfMemoryException. Различие между неограниченным каналом и очередью состоит в том, что когда потребитель пытается получить из пустого канала, он приостанавливается до тех пор, пока какие-либо новые элементы не будут отправлены.

Буферизованный канал

Размер буферизованного канала ограничен указанным числом. Производители могут отправлять элементы в этот канал до тех пор, пока не будет достигнут предел размера. Все элементы хранятся внутри. Когда канал полон, следующий вызов `send` на нём приостанавливается до тех пор, пока не станет доступно больше свободного места.

Канал "Rendezvous"

Канал "Rendezvous" — это канал без буфера, такой же, как буферизованный канал с нулевым размером. Одна из функций (send() или receive()) всегда приостанавливается, пока не будет вызвана другая.

Если функция send() вызывается, и нет приостановленного вызова receive, готового обработать элемент, то send() приостанавливается. Аналогично, если функция receive вызывается, и канал пуст, или, другими словами, нет приостановленного вызова send(), готового отправить элемент, то вызов receive() приостанавливается.

Название "rendezvous" ("встреча в оговоренное время и месте") относится к тому, что send() и receive() должны "встретиться вовремя".

Конфлюэнтный канал

Новый элемент, отправленный в конфлюэнтный канал, перезапишет ранее отправленный элемент, поэтому получатель всегда получит только последний элемент. Вызов send() никогда не приостанавливается.

При создании канала укажите его тип или размер буфера (если вам нужен буферизованный канал):

val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(CONFLATED)
val unlimitedChannel = Channel<String>(UNLIMITED)

По умолчанию создаётся канал "Rendezvous".

В следующем задании вы создадите канал "Rendezvous", две корутины-производителя и корутину-потребителя:

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        channel.send("A1")
        channel.send("A2")
        log("A done")
    }
    launch {
        channel.send("B1")
        log("B done")
    }
    launch {
        repeat(3) {
            val x = channel.receive()
            log(x)
        }
    }
}

fun log(message: Any?) {
    println("[${Thread.currentThread().name}] $message")
}

val channel = Channel<List<User>>()
for (repo in repos) {
    launch {
        val users = TODO()
        // ...
        channel.send(users)
    }
}

repeat(repos.size) {
    val users = channel.receive()
    // ...
}

suspend fun loadContributorsChannels(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) = coroutineScope {

    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    val channel = Channel<List<User>>()
    for (repo in repos) {
        launch {
            val users = service.getRepoContributors(req.org, repo.name)
                .also { logUsers(repo, it) }
                .bodyList()
            channel.send(users)
        }
    }
    var allUsers = emptyList<User>()
    repeat(repos.size) {
        val users = channel.receive()
        allUsers = (allUsers + users).aggregate()
        updateResults(allUsers, it == repos.lastIndex)
    }
}

Тестирование сопроцедур

Теперь давайте протестируем все решения, чтобы проверить, что решение с конкурентными сопроцедурами быстрее, чем решение с функциями suspend, и что решение с каналами быстрее, чем простое решение «прогресс».

В следующей задаче вы сравните общее время выполнения решений. Вы будете имитировать службу GitHub и заставлять эту службу возвращать результаты после заданных таймаутов:

repos request - returns an answer within 1000 ms delay
repo-1 - 1000 ms delay
repo-2 - 1200 ms delay
repo-3 - 800 ms delay

Последовательное решение с функциями suspend должно занимать около 4000 мс (4000 = 1000 + (1000 + 1200 + 800)). Конкурентное решение должно занимать около 2200 мс (2200 = 1000 + max(1000, 1200, 800)).

Для решений, которые показывают прогресс, вы также можете проверить промежуточные результаты с отметками времени.

Соответствующие тестовые данные определены в test/contributors/testData.kt, а файлы Request4SuspendKtTest, Request7ChannelsKtTest и так далее содержат простые тесты, которые используют вызовы имитируемой службы.

Однако здесь есть две проблемы:

• Эти тесты занимают слишком много времени для выполнения. Каждый тест занимает около 2–4 секунд, и вам нужно каждый раз ждать результатов. Это не очень эффективно.

• Вы не можете полагаться на точное время выполнения решения, потому что оно все равно занимает дополнительное время на подготовку и выполнение кода. Вы могли бы добавить константу, но тогда время будет отличаться от машины к машине. Задержки имитируемой службы должны быть больше этой константы, чтобы вы могли увидеть разницу. Если константа составляет 0,5 секунды, то задержки в 0,1 секунды будут недостаточными.

Лучшим способом было бы использовать специальные фреймворки для тестирования времени выполнения, выполняя тот же код несколько раз (что увеличивает общее время еще больше), но это сложно изучить и настроить.

Чтобы решить эти проблемы и убедиться, что решения с заданными задержками тестирования ведут себя как ожидается, одно быстрее другого, используйте виртуальное время с особым диспетчером тестов. Этот диспетчер отслеживает прошедшее виртуальное время с начала и выполняет все немедленно в реальном времени. При запуске сопроцедур на этом диспетчере delay вернется немедленно и переведет виртуальное время.

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

Чтобы использовать виртуальное время, замените вызов runBlocking на runTest. runTest принимает лямбда-расширение для TestScope в качестве аргумента. Когда вы вызываете delay в функции suspend внутри этого специального области действия, delay увеличит виртуальное время вместо задержки в реальном времени:

@Test
fun testDelayInSuspend() = runTest {
    val realStartTime = System.currentTimeMillis() 
    val virtualStartTime = currentTime
        
    foo()
    println("${System.currentTimeMillis() - realStartTime} ms") // ~ 6 ms
    println("${currentTime - virtualStartTime} ms")             // 1000 ms
}

suspend fun foo() {
    delay(1000)    // auto-advances without delay
    println("foo") // executes eagerly when foo() is called
}

Текущее виртуальное время можно проверить, используя свойство currentTime объекта TestScope.

Фактическое время выполнения в этом примере составляет несколько миллисекунд, тогда как виртуальное время равно аргументу задержки, которое равно 1000 миллисекундам.

Чтобы получить полный эффект «виртуального» delay в дочерних сопроцедурах, запустите все дочерние сопроцедуры с помощью TestDispatcher. В противном случае это не сработает. Этот диспетчер автоматически наследуется от других TestScope, если вы не укажете другой диспетчер:

@Test
fun testDelayInLaunch() = runTest {
    val realStartTime = System.currentTimeMillis()
    val virtualStartTime = currentTime

    bar()

    println("${System.currentTimeMillis() - realStartTime} ms") // ~ 11 ms
    println("${currentTime - virtualStartTime} ms")             // 1000 ms
}

suspend fun bar() = coroutineScope {
    launch {
        delay(1000)    // auto-advances without delay
        println("bar") // executes eagerly when bar() is called
    }
}

Если launch вызывается в контексте Dispatchers.Default в приведенном выше примере, тест завершится неудачно. Вы получите исключение, говорящее о том, что задача еще не завершена.

Вы можете протестировать функцию loadContributorsConcurrent() таким образом только в том случае, если она запускает дочерние сопроцедуры с унаследованным контекстом, не изменяя его с помощью диспетчера Dispatchers.Default.

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

По умолчанию компилятор выводит предупреждения, если вы используете экспериментальный API тестирования. Чтобы подавить эти предупреждения, аннотируйте функцию теста или весь класс, содержащий тесты, с помощью @OptIn(ExperimentalCoroutinesApi::class). Добавьте аргумент компилятора, указывающий компилятору, что вы используете экспериментальный API:

compileTestKotlin {
    kotlinOptions {
        freeCompilerArgs += "-Xuse-experimental=kotlin.Experimental"
    }
}

В проекте, соответствующем этому учебнику, аргумент компилятора уже добавлен в скрипт Gradle.

fun testConcurrent() = runTest {
    val startTime = currentTime
    val result = loadContributorsConcurrent(MockGithubService, testRequestData)
    Assert.assertEquals("Wrong result for 'loadContributorsConcurrent'", expectedConcurrentResults.users, result)
    val totalTime = currentTime - startTime

    Assert.assertEquals(
        "The calls run concurrently, so the total virtual time should be 2200 ms: " +
                "1000 for repos request plus max(1000, 1200, 800) = 1200 for concurrent contributors requests)",
        expectedConcurrentResults.timeFromStart, totalTime
    )
}

fun testChannels() = runTest {
    val startTime = currentTime
    var index = 0
    loadContributorsChannels(MockGithubService, testRequestData) { users, _ ->
        val expected = concurrentProgressResults[index++]
        val time = currentTime - startTime
        Assert.assertEquals(
            "Expected intermediate results after ${expected.timeFromStart} ms:",
            expected.timeFromStart, time
        )
        Assert.assertEquals("Wrong intermediate results after $time:", expected.users, users)
    }
}

Что дальше

• Посмотрите мастер-класс Асинхронное программирование с Kotlin на KotlinConf.

• Узнайте больше об использовании виртуального времени и экспериментального пакета тестирования.