Ответ
В асинхронном Python (с использованием asyncio) эндпоинты не блокируются благодаря модели кооперативной многозадачности, управляемой event loop. Когда корутина встречает операцию ввода-вывода (I/O), она "отдает управление", позволяя loop'у выполнять другие задачи.
Как это работает в DevOps-скриптах или веб-серверах (например, FastAPI):
- Event Loop: Запускается один цикл событий, который управляет всеми корутинами.
awaitи Yield Control: Ключевое словоawaitприостанавливает выполнение текущей корутинны до завершения I/O-операции (запрос к API, БД, файлу) и возвращает управление в loop.- Неблокирующий I/O: Под капотом используются системные вызовы (например,
epoll,kqueue), которые уведомляют loop о готовности данных, без простоя потока.
Пример асинхронного скрипта для мониторинга нескольких эндпоинтов одновременно:
import aiohttp
import asyncio
async def check_endpoint(session, url):
try:
async with session.get(url, timeout=5) as resp:
status = resp.status
# В момент ожидания ответа event loop может выполнять другие check_endpoint
return url, status == 200
except Exception as e:
return url, False
async def main():
urls = [
'http://api.service1/health',
'http://api.service2/health',
'http://prometheus:9090/-/healthy'
]
async with aiohttp.ClientSession() as session:
tasks = [check_endpoint(session, url) for url in urls]
# Все запросы выполняются конкурентно, не блокируя друг друга
results = await asyncio.gather(*tasks)
for url, is_healthy in results:
print(f"{url}: {'OK' if is_healthy else 'FAIL'}")
if __name__ == "__main__":
asyncio.run(main())
Это позволяет одному процессу эффективно обрабатывать тысячи одновременных подключений, что критично для высоконагруженных API-гейтвеев или скриптов агрегации метрик.