Тема, шрифт, мова та інші налаштування інтерфейсу.
Статистика використання. Вимкнена, доки ви її не дозволите.
Реклама та вимірювання реклами. Вимкнені, доки ви їх не дозволите.
ПРО САЙТ
Усі відкриті вкладки сайту
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!'`.
Напишіть асинхронний генератор `async_range(start: int, stop: int, delay: float)`, що видає цілі числа від `start` до `stop-1`, роблячи паузу `delay` секунд між кожним значенням. Використайте `async for` для споживання та виводу значень.
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())
Функція `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 с
Щоб тренувати сліпий друк, не підглядайте на фізичну клавіатуру.