import asyncio

async def fetch_data(url: str) -> dict:
    print(f"开始获取: {url}")
    await asyncio.sleep(1)  # 模拟 I/O 操作
    return {"url": url, "data": "response"}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

async def fetch_multiple():
    # 串行执行（慢）
    result1 = await fetch_data("https://api1.example.com")
    result2 = await fetch_data("https://api2.example.com")
    
    # 并发执行（快）
    result1, result2 = await asyncio.gather(
        fetch_data("https://api1.example.com"),
        fetch_data("https://api2.example.com")
    )
    
    return [result1, result2]

import aiohttp
import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        data = await response.json()
        return {"url": url, "status": response.status, "data": data}

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# 使用
urls = [
    "https://api.example.com/users",
    "https://api.example.com/posts",
    "https://api.example.com/comments"
]

results = asyncio.run(fetch_all(urls))

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def get_user(user_id: str) -> dict:
    await asyncio.sleep(0.1)  # 模拟数据库查询
    return {"id": user_id, "name": "Alice"}

async def get_orders(user_id: str) -> list:
    await asyncio.sleep(0.2)  # 模拟数据库查询
    return [{"id": "1", "total": 100}]

@app.get("/api/dashboard/{user_id}")
async def dashboard(user_id: str):
    # 并发获取数据
    user, orders = await asyncio.gather(
        get_user(user_id),
        get_orders(user_id)
    )
    
    return {
        "user": user,
        "orders": orders,
        "order_count": len(orders)
    }

import asyncio
from dataclasses import dataclass

@dataclass
class Task:
    id: str
    payload: dict

async def worker(queue: asyncio.Queue, worker_id: int):
    while True:
        task = await queue.get()
        try:
            print(f"Worker {worker_id} 处理任务: {task.id}")
            await process_task(task)
        finally:
            queue.task_done()

async def process_task(task: Task):
    await asyncio.sleep(0.5)  # 模拟处理
    print(f"任务完成: {task.id}")

async def main():
    queue: asyncio.Queue[Task] = asyncio.Queue()
    
    # 启动 5 个 worker
    workers = [
        asyncio.create_task(worker(queue, i))
        for i in range(5)
    ]
    
    # 添加任务
    for i in range(20):
        await queue.put(Task(id=f"task-{i}", payload={"index": i}))
    
    # 等待所有任务完成
    await queue.join()
    
    # 取消 worker
    for w in workers:
        w.cancel()

asyncio.run(main())

async def fetch_with_limit(urls: list[str], limit: int = 10):
    semaphore = asyncio.Semaphore(limit)
    
    async def fetch_limited(session: aiohttp.ClientSession, url: str):
        async with semaphore:
            return await fetch_url(session, url)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_limited(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# 限制同时最多 10 个并发请求
results = asyncio.run(fetch_with_limit(urls, limit=10))

async def fetch_with_timeout(url: str, timeout: float = 5.0):
    try:
        async with asyncio.timeout(timeout):
            return await fetch_data(url)
    except TimeoutError:
        print(f"请求超时: {url}")
        return None

async def cancellable_task():
    task = asyncio.create_task(long_running_operation())
    
    # 3 秒后取消
    await asyncio.sleep(3)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

class AsyncDatabaseConnection:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

# 使用
async def query_db():
    async with AsyncDatabaseConnection() as conn:
        result = await conn.execute("SELECT * FROM users")
        return result

async def async_range(start: int, stop: int):
    for i in range(start, stop):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for i in async_range(0, 10):
        print(i)

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3
) -> dict:
    for attempt in range(max_retries):
        try:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # 指数退避
            await asyncio.sleep(wait_time)

import signal

async def graceful_shutdown(loop: asyncio.AbstractEventLoop):
    print("收到关闭信号，正在优雅关闭...")
    
    # 取消所有任务
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in tasks:
        task.cancel()
    
    await asyncio.gather(*tasks, return_exceptions=True)
    loop.stop()

async def main():
    loop = asyncio.get_running_loop()
    
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(
            sig,
            lambda: asyncio.create_task(graceful_shutdown(loop))
        )
    
    # 主逻辑
    await run_server()