`asyncio` — Cooperative Concurrency
Python · Concurrency · Run with python3
Explanation
`asyncio` runs many coroutines on a single thread using cooperative multitasking. A coroutine (defined with `async def`) yields control at `await` points. The event loop drives execution, switching between coroutines whenever one awaits I/O. `asyncio.run()` is the entry point. `asyncio.gather()` runs multiple coroutines concurrently. `asyncio.sleep(0)` yields control without actually sleeping. All I/O must use `async`-aware libraries (`aiohttp`, `asyncpg`, `aiofiles`) — calling blocking I/O inside a coroutine blocks the entire event loop.
Real-World Scenario
A high-throughput API server handling 10,000 simultaneous connections: `asyncio` interleaves waiting coroutines on one thread with near-zero memory per "connection" compared to spawning a thread per connection. ```python import asyncio import random async def fetch(session_id: int, url: str) -> dict: """Simulates an async HTTP request.""" await asyncio.sleep(random.uniform(0.05, 0.3)) return {"session": session_id, "url": url, "status": 200} async def main() -> None: urls = [f"https://api.example.com/v1/item/{i}" for i in range(8)] # Run all concurrently, collect results tasks = [fetch(i, url) for i, url in enumerate(urls)] results = await asyncio.gather(*tasks) for r in results: print(f"[{r['session']}] {r['url']} → {r['status']}") print(f"\nCompleted {len(results)} requests") asyncio.run(main()) ```
Quick Snippet
import asyncio
import random
async def fetch(session_id: int, url: str) -> dict:
"""Simulates an async HTTP request."""
await asyncio.sleep(random.uniform(0.05, 0.3))
return {"session": session_id, "url": url, "status": 200}
async def main() -> None:
urls = [f"https://api.example.com/v1/item/{i}" for i in range(8)]
# Run all concurrently, collect results
tasks = [fetch(i, url) for i, url in enumerate(urls)]
results = await asyncio.gather(*tasks)
for r in results:
print(f"[{r['session']}] {r['url']} → {r['status']}")
print(f"\nCompleted {len(results)} requests")
asyncio.run(main())Full Example
# Compile: python3 asyncio_basics.py
# Output:
# Completed 8 requests
import asyncio
import random
async def fetch(session_id: int, url: str) -> dict:
await asyncio.sleep(random.uniform(0.01, 0.05))
return {"session": session_id, "url": url, "status": 200}
async def main() -> None:
urls = [f"https://api.example.com/v1/item/{i}" for i in range(8)]
tasks = [fetch(i, url) for i, url in enumerate(urls)]
results = await asyncio.gather(*tasks)
for r in results:
print(f"[{r['session']}] status={r['status']}")
print(f"\nCompleted {len(results)} requests")
asyncio.run(main())
python3 asyncio_basics.py