Ответ
Да, у меня есть практический опыт асинхронного программирования на Python с использованием asyncio и сопутствующих библиотек (aiohttp, aiomysql, asyncpg). Я применял его для создания высокопроизводительных сетевых сервисов, где необходимо эффективно обрабатывать множество одновременных I/O-операций, таких как запросы к API, работа с базами данных или файловой системой.
Пример: параллельный сбор данных с нескольких REST API эндпоинтов.
import asyncio
import aiohttp
from typing import List, Dict
async def fetch_json(session: aiohttp.ClientSession, url: str) -> Dict:
"""Асинхронно выполняет GET-запрос и возвращает JSON."""
try:
async with session.get(url, timeout=10) as response:
response.raise_for_status()
return await response.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
print(f"Ошибка при запросе к {url}: {e}")
return {}
async def fetch_all_urls(urls: List[str]) -> List[Dict]:
"""Создает задачи для всех URL и собирает результаты."""
async with aiohttp.ClientSession() as session:
tasks = [fetch_json(session, url) for url in urls]
# Выполняем все задачи конкурентно и ждем их завершения
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if isinstance(r, dict)]
# Использование
async def main():
api_endpoints = [
'https://api.example.com/users',
'https://api.example.com/posts',
'https://api.example.com/comments'
]
data = await fetch_all_urls(api_endpoints)
print(f"Получено {len(data)} ответов")
# Запуск event loop
if __name__ == '__main__':
asyncio.run(main())
Ключевые навыки и понимание:
- Работа с корутинами (coroutines), задачами (Tasks) и циклом событий (Event Loop).
- Использование
async/awaitсинтаксиса для написания неблокирующего кода. - Понимание конкурентности и параллелизма, управление состоянием гонки (race conditions) с помощью примитивов (
asyncio.Lock,asyncio.Semaphore). - Знание ограничений: асинхронность не ускоряет CPU-bound задачи, для них нужны процессы (
multiprocessing).