Статистика использования. Отключена, пока вы её не разрешите.
Реклама и измерение рекламы. Отключены, пока вы их не разрешите.
О САЙТЕ
Все открытые вкладки сайта
Python · Синтаксис · Продвинутый уровень
Асинхронный Python
10 задач
Пишите конкурентные программы с `async def`, `await` и `asyncio`. Охватывает корутины, `asyncio.gather`, задачи, `asyncio.Queue` и асинхронные итераторы.
Краткое введение в тему и пояснения перед упражнениями (упражнения ниже):
**Зачем async? Проблема блокирующего I/O**
Обычный сервер большую часть времени *ждёт* — ответа базы данных, HTTP-запроса, чтения файла. При синхронном коде поток заблокирован и не может ничего делать.
Async решает это позволяя одному потоку обрабатывать множество ожидающих задач одновременно. Пока одна задача ждёт I/O, цикл событий запускает другую.
**Чем async НЕ является**
Async не ускоряет CPU-bound работу. Если код тратит время на вычисления (сортировка, обработка изображений), async ничего не даст — поток занят, а не ждёт. Для CPU-работы используйте `multiprocessing`.
**Корутины: основная единица**
Корутина — функция определённая с `async def`. Вызов её НЕ запускает тело — она создаёт объект корутины:
```python
import asyncio
async def fetch(url: str) -> str:
await asyncio.sleep(1) # передаёт управление циклу событий
return f'data from {url}'
# Ничего не делает сам по себе:
coro = fetch('https://example.com') # объект корутины
# Запустить:
result = asyncio.run(fetch('https://example.com'))
```
**`await` — приостановка и возобновление**
`await expr` приостанавливает текущую корутину и передаёт управление циклу событий. Когда ожидаемое завершается, цикл возобновляет корутину с места паузы.
```python
async def slow_task(name: str, delay: float):
print(f'{name}: starting')
await asyncio.sleep(delay)
print(f'{name}: done after {delay}s')
async def main():
# Последовательно — итого 3 секунды:
await slow_task('A', 1)
await slow_task('B', 2)
asyncio.run(main())
```
**`asyncio.gather` — параллельное выполнение задач**
`gather` планирует несколько корутин параллельно в одном цикле событий:
```python
async def main():
# Конкурентно — итого ~2 секунды:
results = await asyncio.gather(
slow_task('A', 1),
slow_task('B', 2),
)
asyncio.run(main())
```
**`asyncio.Task` vs корутина**
Корутина выполняется только когда её awaited. `Task` оборачивает корутину и планирует её выполнение *независимо* в цикле событий:
```python
import asyncio
async def background_job():
await asyncio.sleep(5)
print('background done')
async def main():
task = asyncio.create_task(background_job()) # планируется сразу
print('doing other work') # выполняется сейчас
await asyncio.sleep(1)
await task # ждём завершения
asyncio.run(main())
```
**`async for` и `async with`**
```python
# async with — async контекстный менеджер
async with aiofiles.open('data.txt') as f:
content = await f.read()
# async for — async итератор
async for line in aiohttp_response.content:
process(line)
# async генератор:
async def lines(filename):
async with aiofiles.open(filename) as f:
async for line in f:
yield line.strip()
```
**Таймаут и отмена**
```python
import asyncio
async def main():
try:
result = await asyncio.wait_for(slow(), timeout=2.0)
except asyncio.TimeoutError:
print('timed out')
# Python 3.11+:
async with asyncio.timeout(2.0):
result = await slow()
```
**Реальный паттерн: конкурентные HTTP-запросы**
```python
import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
return await asyncio.gather(*tasks)
results = asyncio.run(fetch_all(['https://example.com', 'https://example.org']))
# Оба загружаются конкурентно — время = самый медленный, не сумма
```
Частые ошибки, asyncio vs потоки vs multiprocessing, справочник
Напишите корутину `async def greet(name: str)`, которая выводит `'Привет, <name>!'`, ожидает 0.1 секунды через `asyncio.sleep`, затем выводит `'До свидания, <name>!'`. Запустите её через `asyncio.run(greet('Алиса'))`.
Напишите корутину `fetch(url: str, delay: float) -> str`, симулирующую HTTP-запрос: ждёт `delay` секунд и возвращает `f'Response from {url}'`. Используйте `asyncio.gather` для запуска трёх fetch конкурентно: `fetch('A', 0.3)`, `fetch('B', 0.1)`, `fetch('C', 0.2)`. Выведите все результаты. Общее время — около 0.3с, а не 0.6с.
import asyncio
async def fetch(url: str, delay: float) -> str:
await asyncio.sleep(delay)
return f'Response from {url}'
async def main() -> None:
results = await asyncio.gather(
fetch('A', 0.3),
fetch('B', 0.1),
fetch('C', 0.2),
)
for r in results:
print(r)
asyncio.run(main())
# Response from A
# Response from B
# Response from C
Напишите корутину `main()`, создающую два фоновых задания через `asyncio.create_task`: одно считает от 3 до 1 (выводит каждое число с задержкой 0.1с), другое выводит `'Работаем...'` три раза с задержкой 0.15с. Дождитесь обоих заданий в конце. Обратите внимание, что вывод чередуется.
import asyncio
async def countdown() -> None:
for i in range(3, 0, -1):
print(f'Обратный отсчёт: {i}')
await asyncio.sleep(0.1)
async def worker() -> None:
for _ in range(3):
print('Работаем...')
await asyncio.sleep(0.15)
async def main() -> None:
# создайте задания и дождитесь их
pass
asyncio.run(main())
Решение
import asyncio
async def countdown() -> None:
for i in range(3, 0, -1):
print(f'Обратный отсчёт: {i}')
await asyncio.sleep(0.1)
async def worker() -> None:
for _ in range(3):
print('Работаем...')
await asyncio.sleep(0.15)
async def main() -> None:
task1 = asyncio.create_task(countdown())
task2 = asyncio.create_task(worker())
await task1
await task2
asyncio.run(main())
Реализуйте паттерн производитель-потребитель с `asyncio.Queue`. Корутина `producer` кладёт числа 1–5 в очередь (задержка 0.05с между числами), затем кладёт `None` как сигнал завершения. Корутина `consumer` читает элементы из очереди до появления `None`, выводя `'Consuming: <n>'` для каждого числа.
Напишите корутину `slow_op(seconds: float) -> str`, ожидающую `seconds` секунд и возвращающую `'done'`. Используйте `asyncio.wait_for(slow_op(5), timeout=0.2)` для попытки операции с таймаутом 0.2 секунды. Перехватите `asyncio.TimeoutError` и выведите `'Operation timed out!'`.
import asyncio
async def async_range(start: int, stop: int, delay: float):
for i in range(start, stop):
yield i
await asyncio.sleep(delay)
async def main() -> None:
async for value in async_range(0, 4, 0.05):
print(value)
asyncio.run(main())
Создайте класс `AsyncTimer`, работающий как асинхронный контекстный менеджер. `__aenter__` фиксирует время начала и выводит `'Timer started'`. `__aexit__` вычисляет прошедшее время и выводит `f'Elapsed: {elapsed:.2f}s'`. Используйте с `async with AsyncTimer():`.
import asyncio
import time
class AsyncTimer:
async def __aenter__(self):
pass
async def __aexit__(self, *args):
pass
async def main() -> None:
async with AsyncTimer():
await asyncio.sleep(0.1)
asyncio.run(main())
# Timer started
# Elapsed: 0.10s
У вас есть 8 задач, но одновременно должно выполняться не более 3. Используйте `asyncio.Semaphore(3)` для ограничения конкурентности. Каждая задача выводит `'Task N started'`, ждёт 0.1с, затем `'Task N done'`. Запустите все 8 задач через `asyncio.gather`.
import asyncio
async def task(n: int, sem: asyncio.Semaphore) -> None:
async with sem:
print(f'Task {n} started')
await asyncio.sleep(0.1)
print(f'Task {n} done')
async def main() -> None:
sem = asyncio.Semaphore(3)
# создайте и запустите 8 задач (n от 1 до 8)
pass
asyncio.run(main())
Решение
import asyncio
async def task(n: int, sem: asyncio.Semaphore) -> None:
async with sem:
print(f'Task {n} started')
await asyncio.sleep(0.1)
print(f'Task {n} done')
async def main() -> None:
sem = asyncio.Semaphore(3)
await asyncio.gather(*(task(n, sem) for n in range(1, 9)))
asyncio.run(main())
09
Выполнение блокирующего кода через run_in_executor
Функция `blocking_sum(n)` вычисляет сумму 1..n через цикл (симулирует медленную работу). Оберните её в асинхронную `async_sum(n)`, используя `loop.run_in_executor(None, blocking_sum, n)`, чтобы не блокировать цикл событий. Запустите два вызова `async_sum` конкурентно через `asyncio.gather`.
Используйте `asyncio.Event` для координации двух корутин. Корутина `setter` ждёт 0.15с, выводит `'Setting event'`, затем устанавливает событие. Корутина `waiter` выводит `'Waiting for event...'`, ждёт через `await event.wait()`, затем выводит `'Event received!'`. Запустите обе через `asyncio.gather`.
Мы используем необходимые cookies для работы сайта. С вашего разрешения мы также можем сохранять настройки сайта и использовать аналитические и рекламные cookies, чтобы понимать использование сайта и поддерживать развитие проекта.
* Вы всегда можете изменить свой выбор в настройках сайта.
Выберите категории cookies
Настройки аналитики
Можно отключить аналитику использования платформы. Также можно отправить в Google Analytics запрос на удаление данных об использовании этого сайта, связанных с этим браузером.
Учебный workspace
Учитесь, читая, запуская код и решая задачи.
Практикуйте программирование с пояснениями тем, упражнениями, инструментами browser IDE, проверкой regex и тренировкой печати кода в одном workspace.
Открывайте инструменты во вкладках.Упражнения, IDE-инструменты и тренажеры остаются доступными как вкладки сайта.
Переключайтесь без потери контекста.Переходите между пояснениями, кодом и утилитами, сохраняя свое место.
Используйте sidebar как карту.Левые панели содержат навигацию, настройки, файлы, libraries и управление инструментами.
PythonJavaScriptSQLite
Одна IDE, три практичных режима
Python в браузере.Запускайте небольшие скрипты, пробуйте библиотеки и тренируйте API-запросы без установки.
JavaScript для быстрых экспериментов.Проверяйте код для браузера и сравнивайте идеи рядом с учебными материалами.
SQLite для практики с данными.Открывайте обозреватель базы данных, изучайте таблицы, пишите запросы и учитесь SQL локально.
ТемаIDE
Работайте рядом в split tabs
Держите инструкции перед глазами.Откройте упражнение или справочную страницу рядом с IDE, вместо постоянных переключений.
Сравнивайте инструменты во время обучения.Размещайте проверки regex, пояснения и эксперименты с кодом рядом, когда это нужно для задачи.
Закройте split, когда закончите.Workspace вернётся к одной сфокусированной вкладке, а открытые вкладки сайта останутся доступны.
Тренажер слепой печати кода
Или просто текста
Тренажер рассчитан на физическую клавиатуру.Откройте этот раздел на ноутбуке или компьютере с широким экраном. На телефоне тренировка слепой печати не будет корректной.
Скорость: 0 зн/мин
0 слов/мин
Лучшая скорость (60с): 0 зн/мин
0 слов/мин
Ошибки: 0
Общее время: 0.0 с
Для активации режима слепого набора не подсматривайте на физическую клавиатуру.