Как отменить выполнение корутины
При работе приложения может сложиться необходимость отменить выполнение корутины. Например, в мобильном приложении запущена корутина для загрузки данных с некоторого интернет-ресуса, но пользователь решил перейти к другой странице приложения, и ему больше не нужны эти данные. В этом случае чтобы зря не тратить ресурсу системы, мы можем предусмотреть отмену выполнения корутины.
Для отмены выполнения корутины у объекта Job может применяться метод cancel() :
import kotlinx.coroutines.* suspend fun main() = coroutineScope < val downloader: Job = launch< println("Начинаем загрузку файлов") for(i in 1..5)< println("Загружен файл $i") delay(500L) >> delay(800L) // установим задержку, чтобы несколько файлов загрузились println("Надоело ждать, пока все файлы загрузятся. Прерву-ка я загрузку. ") downloader.cancel() // отменяем корутину downloader.join() // ожидаем завершения корутины println("Работа программы завершена") >
В данном случае определена корутина, которая имитирует загрузку файлов. В цикле пробегаемся от 1 до 5 и условно загружаем пять файлов.
Далее вызов метода downloader.cancel() сигнализирует корутине, что надо прервать выполнение. Затем с помощью метода join() ожидаем завершения корутина, которая прервана. В итоге получим консольный вывод наподобие следующего:
Начинаем загрузку файлов Загружен файл 1 Загружен файл 2 Надоело ждать, пока все файлы загрузятся. Прерву-ка я загрузку. Работа программы завершена
Также вместо двух методов cancel() и join() можно использовать один сборный метод cancelAndJoin() :
import kotlinx.coroutines.* suspend fun main() = coroutineScope < val downloader: Job = launch< println("Начинаем загрузку файлов") for(i in 1..5)< println("Загружен файл $i") delay(500L) >> delay(800L) println("Надоело ждать, пока все файлы загрузятся. Прерву-ка я загрузку. ") downloader.cancelAndJoin() // отменяем корутину и ожидаем ее завершения println("Работа программы завершена") >
Обработка исключения CancellationException
Все suspend-функции в пакете kotlinx.coroutines являются прерываемыми (cancellable). Это значит, что они проверяют, прервана ли корутина. И если ее выполнение прервано, они генерируют исключение типа CancellationException . И в самой корутине мы можем перехватить это исключение, чтобы обработать отмену корутины. Например:
import kotlinx.coroutines.* suspend fun main() = coroutineScope < val downloader: Job = launch< try < println("Начинаем загрузку файлов") for(i in 1..5)< println("Загружен файл $i") delay(500L) >> catch (e: CancellationException ) < println("Загрузка файлов прервана") >finally < println("Загрузка завершена") >> delay(800L) println("Надоело ждать. Прерву-ка я загрузку. ") downloader.cancelAndJoin() // отменяем корутину и ожидаем ее завершения println("Работа программы завершена") >
Здесь код выполнения корутины обернут в конструкцию try . Если корутина будет прервана извне, то с помощью блока catch и перехвата исключения CancellationException мы сможем обработать отмену корутины.
И если нам надо выполнить некоторые завершающие действия, например, освободить используемые в корутине ресурсы — закрыть файлы, различные подключения к внешним ресурсам, то это можно сделать в блоке finally . Но в данном случае в этом блоке просто выводим диагностическое сообщение.
В итоге при вызове метода downloader.cancel() производейт отмена корутины. Будет сгенерировано исключение, и в корутине в блоке catch мы сможем ее обработать. В итоге получим следующий консольный вывод:
Начинаем загрузку файлов Загружен файл 1 Загружен файл 2 Надоело ждать. Прерву-ка я загрузку. Загрузка файлов прервана Загрузка завершена Работа программы завершена
Отмена выполнения async-корутины
Подобным образом можно отменять выполнение и корутин, создаваемых с помощью функции async() . В этом случае обычно вызов метода await() помещается в блок try:
import kotlinx.coroutines.* suspend fun main() = coroutineScope < // создаем и запускаем корутину val message = async < getMessage() >// отмена корутины message.cancelAndJoin() try < // ожидаем получение результата println("message: $") > catch (e:CancellationException) < println("Coroutine has been canceled") >println(«Program has finished») > suspend fun getMessage() : String
Консольный вывод программы:
Coroutine has been canceled Program has finished
Отмена корутин и тайм-ауты
В этом разделе рассматривается отмена корутин и тайм-ауты.
Отмена выполнения корутин
В долго работающем приложении вам может понадобиться детальное управление фоновыми корутинами. Например, пользователь может закрыть страницу, которая запускала корутину, из-за чего её результат больше не нужен, и её действие можно отменить. Функция launch возвращает Job , которую можно использовать для отмены запущенной корутины.
val job = launch < repeat(1000) < i ->println("job: I'm sleeping $i . ") delay(500L) > > delay(1300L) // delay a bit println("main: I'm tired of waiting!") job.cancel() // cancels the job job.join() // waits for job's completion println("main: Now I can quit.") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-01.kt). —>
Этот код выведет следующее:
job: I'm sleeping 0 . job: I'm sleeping 1 . job: I'm sleeping 2 . main: I'm tired of waiting! main: Now I can quit. Как только главная функция вызывает job.cancel , мы больше не видим какого-либо вывода с другой корутины, потому что она была отменена. Существует также cancelAndJoin функция-расширение Job , которая объединяет вызовы cancel и join.
Отмена кооперативна
Отмена корутин кооперативна. Код корутины должен взаимодействовать, чтобы его можно было отменить. Все suspend-функции в kotlinx.coroutines — отменяемые. Они проверяют отмену корутины, и в случае отмены выбрасывают исключение CancellationException . Однако, если корутина работает над вычислениями и не проверяет на отмену, то её нельзя отменить, как это происходит, например, здесь:
val startTime = System.currentTimeMillis() val job = launch(Dispatchers.Default) < var nextPrintTime = startTime var i = 0 while (i < 5) < // computation loop, just wastes CPU // print a message twice a second if (System.currentTimeMillis() >= nextPrintTime) < println("job: I'm sleeping $. ") nextPrintTime += 500L > > > delay(1300L) // delay a bit println("main: I'm tired of waiting!") job.cancelAndJoin() // cancels the job and waits for its completion println("main: Now I can quit.") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-02.kt). —>
Запустив этот код, вы увидите как корутина продолжает выводить на экран «I’m sleeping» даже после отмены, пока job не завершится после пяти итераций.
Делаем код с вычислениями отменяемым
Есть два способа сделать вычислительный код отменяемым. Первый – периодически вызвать suspend-функцию, которая проверяет, активна ли корутина. Для этого хорошо подходит функция yield . Другой — явно проверять статус отмены. Попробуем этот подход.
val startTime = System.currentTimeMillis() val job = launch(Dispatchers.Default) < var nextPrintTime = startTime var i = 0 while (isActive) < // cancellable computation loop // print a message twice a second if (System.currentTimeMillis() >= nextPrintTime) < println("job: I'm sleeping $. ") nextPrintTime += 500L > > > delay(1300L) // delay a bit println("main: I'm tired of waiting!") job.cancelAndJoin() // cancels the job and waits for its completion println("main: Now I can quit.") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-03.kt). —>
Как вы можете увидеть, теперь цикл отменяется. isActive — это extension-параметр, доступный внутри корутины, благодаря объекту CoroutineScope .
Закрытие ресурсов при помощи finally
Отменяемые suspend-функции при отмене выбрасывают исключение CancellationException , которое может быть обработано обычным путём. Например, выражение try <. >finally <. >и Kotlin-функция use обыкновенно выполняют свои функции при завершении (отмене) корутин.
val job = launch < try < repeat(1000) < i ->println("job: I'm sleeping $i . ") delay(500L) > > finally < println("job: I'm running finally") >> delay(1300L) // delay a bit println("main: I'm tired of waiting!") job.cancelAndJoin() // cancels the job and waits for its completion println("main: Now I can quit.") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-04.kt). —>
И join , и cancelAndJoin ожидают завершения всех финальных стадий, поэтому приведённый выше пример даёт такой вывод:
job: I'm sleeping 0 . job: I'm sleeping 1 . job: I'm sleeping 2 . main: I'm tired of waiting! job: I'm running finally main: Now I can quit. Запуск неотменяемого блока
Любая попытка использовать suspend-функцию в блоке finally предыдущего примера приводит к CancellationException , потому что корутина, выполняющая этот код, отменена. Обычно это не проблема, так как все нормально работающие операции закрытия (закрытие файла, отмена Job или закрытие любого вида канала) обычно не блокируются и не требуют каких-либо suspend-функций. Однако в редких случаях, когда вам нужно приостановить работу отмененной корутины, вы можете обернуть соответствующий код в withContext(NonCancellable) <. >с использованием функции withContext и контекста NonCancellable , как показано в следующем примере:
val job = launch < try < repeat(1000) < i ->println("job: I'm sleeping $i . ") delay(500L) > > finally < withContext(NonCancellable) < println("job: I'm running finally") delay(1000L) println("job: And I've just delayed for 1 sec because I'm non-cancellable") >> > delay(1300L) // delay a bit println("main: I'm tired of waiting!") job.cancelAndJoin() // cancels the job and waits for its completion println("main: Now I can quit.") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-05.kt). —>
Тайм-аут
Самая очевидная практическая причина отменить выполнение корутины — время её выполнения превысило некоторый тайм-аут. Хотя можно вручную отслеживать обращение к соответствующему Job и запускать отдельную корутину для отмены отслеживаемой после тайм-аута, есть готовая к использованию функция withTimeout , которая делает это. Посмотрите на следующий пример:
withTimeout(1300L) < repeat(1000) < i ->println("I'm sleeping $i . ") delay(500L) > > You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-06.kt). —>
Этот код выведет следующее:
I'm sleeping 0 . I'm sleeping 1 . I'm sleeping 2 . Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1300 ms Исключение TimeoutCancellationException , создаваемое withTimeout , является подклассом CancellationException . Мы никогда раньше не видели его трассировку стека, напечатанную на консоли. Это потому, что внутри отмененной корутины CancellationException считается нормальной причиной её завершения. Однако в этом примере мы использовали withTimeout прямо внутри функции main .
Поскольку отмена является лишь исключением, все ресурсы закрываются в обычном порядке. Вы можете обернуть код с тайм-аутом в блоке try <. >catch (e: TimeoutCancellationException) <. >, если вам нужно сделать какое-то дополнительное действие специально для любого тайм-аута или использовать функцию withTimeoutOrNull . Она похожа на withTimeout , но возвращает null по тайм-ауту вместо создания исключения.
val result = withTimeoutOrNull(1300L) < repeat(1000) < i ->println("I'm sleeping $i . ") delay(500L) > "Done" // will get cancelled before it produces this result > println("Result is $result") You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-07.kt). —>
Теперь ошибки при выполнении этой корутины не будет:
I'm sleeping 0 . I'm sleeping 1 . I'm sleeping 2 . Result is null Асинхронный тайм-аут и ресурсы
Событие тайм-аута в withTimeout является асинхронным по отношению к коду, работающему в его блоке, и может произойти в любое время, даже прямо перед возвратом из блока тайм-аута. Имейте это в виду, если вы открываете или приобретаете ресурс внутри блока, который необходимо закрыть или освободить за пределами блока.
Например, здесь мы имитируем закрываемый ресурс с помощью класса Resource , который просто отслеживает, сколько раз он был создан путем увеличения счетчика acquired и уменьшения этого счетчика из его функции close . Давайте запустим много корутин с небольшим таймаутом, попробуем получить этот ресурс изнутри блока withTimeout после небольшой задержки и освободить его извне.
var acquired = 0 class Resource < init < acquired++ >// Acquire the resource fun close() < acquired-- >// Release the resource > fun main() < runBlocking < repeat(100_000) < // Launch 100K coroutines launch < val resource = withTimeout(60) < // Timeout of 60 ms delay(50) // Delay for 50 ms Resource() // Acquire a resource and return it from withTimeout block >resource.close() // Release the resource > > > // Outside of runBlocking all coroutines have completed println(acquired) // Print the number of resources still acquired > You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-08.kt). —>
Если вы запустите приведенный выше код, вы увидите, что он не всегда выводит ноль, хотя это может зависеть от таймингов вашей машины, вам может потребоваться настроить тайм-ауты в этом примере, чтобы действительно увидеть ненулевые значения.
Note, that incrementing and decrementing `acquired` counter here from 100K coroutines is completely safe, > since it always happens from the same main thread. More on that will be explained in the chapter > on coroutine context. —>
Обратите внимание, что увеличение и уменьшение счетчика acquired здесь из 100 000 корутин совершенно безопасно, так как это всегда происходит из одного и того же основного потока. Подробнее об этом будет рассказано в главе о контексте корутин.
Чтобы обойти эту проблему, вы можете сохранить ссылку на ресурс в переменной, а не возвращать ее из блока withTimeout .
runBlocking < repeat(100_000) < // Launch 100K coroutines launch < var resource: Resource? = null // Not acquired yet try < withTimeout(60) < // Timeout of 60 ms delay(50) // Delay for 50 ms resource = Resource() // Store a resource to the variable if acquired >// We can do something else with the resource here > finally < resource?.close() // Release the resource if it was acquired >> > > // Outside of runBlocking all coroutines have completed println(acquired) // Print the number of resources still acquired You can get the full code [here](../../kotlinx-coroutines-core/jvm/test/guide/example-cancel-09.kt). —>
Этот пример всегда выводит 0, а ресурсы не утекают.
Отмена Retrofit Request. Coroutines
К примеру, выполняется какой-либо запрос на сервер с помощью Retrofit 2 и Coroutines. В это время (время выполнения запроса), пользователь смещается на предыдущий фрагмент. С помощью чего можно отменить запрос, который был запущен с помощью Retrofit? В какую сторону копать? Или же, пользователь скрыл приложение, каким образом можно продолжить выполнение запроса и всего того, что он должен выполнить? Что для этого необходимо?
Отслеживать
задан 11 июл 2020 в 23:06
63 1 1 серебряный знак 7 7 бронзовых знаков
1 ответ 1
Сортировка: Сброс на вариант по умолчанию
Для того, чтоб использовать корутины вы создаете объект CoroutineScope . Что-то вроде такого
private val parentJob: Job = Job() private val backgroundScope: CoroutineScope = CoroutineScope(parentJob + Dispatchers.IO) и потом запускаете:
backgroundScope.launch < // some action >Если пользователь к примеру покинул экран и вам больше не нужно выполнять это действие то достаточно всего-лишь вызвать метод cancel() на вашем scope .
backgroundScope.cancel()
Так же вы можете вызвать метод cancel() и на самом объекте Job тем самым завершив все CoroutineScope которые были созданы внутри объекта Job Посмотрите эту лекцию по корутинам. Не пожалеете.
Как правильно остановить корутину
Корутина работает благодаря тому, что event loop раз за разом вызывает её метод coroutine.send(…) . Если перестать вызывать send , то и корутина перестанет работать. Это простой и изящный способ остановить корутину, но не все так просто.
Рассмотрим пример. У нас есть компьютерная игра, которая хранит в базе данных достижения игрока. Функция работает всю игру, и каждые 5 секунд сохраняет в БД список достижений:
async def save_achivements(user_id): connection = open_connection() while True: achivements = get_achivements(user_id) await connection.send(key=user_id, value=achivements) await asyncio.sleep(5) connection.close() С этим кодом есть проблема. Если случится ошибка внутри while True: , то до закрытия соединения с БД дело так и не дойдет. Возникнет исключение, нормальный поток исполнения команд прервется и connection.close() никто не вызовет. От такой проблемы спасает try finally :
async def save_achivements(user_id): connection = open_connection() try: while True: achivements = get_achivements(user_id) await connection.send(key=user_id, value=achivements) await asyncio.sleep(5) finally: connection.close() Теперь в случае ошибки сработает код внутри finally и соединение будет закрыто вызовом connection.close() , ни смотря ни на что. Это победа!
А что произойдет, если корутину save_achivements(…) просто перестанут вызывать? Игрок прервал игру, корутина save_achivements(…) перестала быть нужной, поэтому её выкинули из event loop и соединение осталось незакрытым. Простой и элегантный способ остановки корутины сломал нам try finally .
Из-за этой проблемы у корутин есть еще один метод — coroutine.throw(…) . Он принимает объект исключения — coroutine.throw(exc_obj) — и забрасывает его внутрь корутины. Исключение всплывает по стеку вызовов корутин, как в обычных синхронных функциях.
CancelledError
Event loop, встроенный в библиотеку asyncio никогда не прерывает работу корутин, вместо этого он просит их остановиться самим. Его просьба — это исключение CancelledError . Его можно перехватить и обработать, как обычное исключение. Воспользуемся этим фактом, чтобы перед выходом из игры записать в БД дату последнего сохранения:
import time import asyncio async def save_achivements(user_id): connection = open_connection() try: while True: achivements = get_achivements(user_id) await connection.send(key=user_id, value=achivements) await asyncio.sleep(5) except asyncio.CancelledError: timestamp = time.time() await connection.send(key=user_id, value=achivements) await connection.send(key='last_update', value=timestamp) # отпускаем перехваченный CancelledError raise finally: connection.close() Внутри asyncio even loop происходит примерно следующее:
coroutine = save_achivements(user_id) coroutine.send(...) coroutine.send(...) ... coroutine.throw(asyncio.CancelledError()) coroutine.send(...) coroutine.send(...) ...
Обратите внимание, что CancelledError не заблокировал работу event loop. Метод coroutine.send(…) продолжает вызываться, авейты исправно работают, а значит, завершение работы корутины тоже может быть асинхронным с вызовами await connection.send(…) .
Никто не может остановить корутину, пока та сама не пожелает. Внешний код будет терпеливо ждать, когда же корутина закончит работу и, чтобы он узнал об этом, в конце отпускаем перехваченный CancelledError . Event loop asyncio поймает это исключение и уведомит всех, кто ждал этого завершения. Подробнее читайте в документации по asyncio.Task.
Остановить корутину может только сама корутина
StopIteration vs CancelledError
StopIteration случается, когда корутина уже дошла до return и успела завершить свою работу. Это исключение нельзя перехватить внутри корутины, оно существует только снаружи.
CancelledError — это исключение просит корутину об остановке. Его специально пробрасывают внутрь, и его можно перехватить. От появления CancelledError до фактического завершения работы корутины может пройти много времени, это остается на её усмотрение.
CancelledError просит об остановке, а StopIteration — её констатирует
Попробуйте бесплатные уроки по Python
Получите крутое код-ревью от практикующих программистов с разбором ошибок и рекомендациями, на что обратить внимание — бесплатно.
Переходите на страницу учебных модулей «Девмана» и выбирайте тему.