# 協(xié)程示例：簡(jiǎn)單的生產(chǎn)者-消費(fèi)者模型
def producer():
    """生產(chǎn)者：生成數(shù)據(jù)"""
    for i in range(5):
        print(f"生產(chǎn): {i}")
        yield i  # 暫停并返回值
def consumer():
    """消費(fèi)者：處理數(shù)據(jù)"""
    received = []
    while len(received) < 5:
        # 從生成器獲取值
        item = next(coroutine)
        received.append(item)
        print(f"消費(fèi): {item}")
    return received
# 創(chuàng)建協(xié)程
coroutine = producer()
# 運(yùn)行協(xié)程直到它yield
result = consumer()
print(f"最終結(jié)果: {result}")

# 早期風(fēng)格的協(xié)程（使用send）
def coro():
    """使用send的協(xié)程"""
    print("協(xié)程開始")
    value = yield "初始值"  # yield可以返回值
    print(f"收到值: {value}")
    value = yield f"處理: {value}"
    print(f"收到值: {value}")
    yield "結(jié)束"
c = coro()
# 首先需要send(None)或next()啟動(dòng)協(xié)程
result1 = c.send(None)  # 或 next(c)
print(f"第1個(gè)yield值: {result1}")
result2 = c.send("你好")
print(f"第2個(gè)yield值: {result2}")
result3 = c.send("再見")
print(f"第3個(gè)yield值: {result3}")

# 使用async/await定義協(xié)程
async def fetch_data():
    """模擬異步獲取數(shù)據(jù)"""
    print("開始獲取數(shù)據(jù)...")
    await asyncio.sleep(1)  # 模擬I/O操作
    print("數(shù)據(jù)獲取完成")
    return {"data": "result", "status": 200}
async def process_data(data):
    """模擬異步處理數(shù)據(jù)"""
    print("開始處理數(shù)據(jù)...")
    await asyncio.sleep(0.5)
    print("數(shù)據(jù)處理完成")
    return data.upper()
async def main():
    """主協(xié)程"""
    # 創(chuàng)建協(xié)程任務(wù)
    data = await fetch_data()
    result = await process_data(data['data'])
    print(f"最終結(jié)果: {result}")
# 運(yùn)行協(xié)程
asyncio.run(main())

# 事件循環(huán)的基本使用
import asyncio
def hello():
    """同步函數(shù)"""
    print("Hello")
async def async_hello():
    """異步函數(shù)"""
    print("Async Hello")
    await asyncio.sleep(0)
    print("Async Hello again")
# 方法1：使用asyncio.run（推薦）
asyncio.run(async_hello())
# 方法2：手動(dòng)創(chuàng)建事件循環(huán)
loop = asyncio.new_event_loop()
try:
    loop.run_until_complete(async_hello())
finally:
    loop.close()
# 方法3：獲取當(dāng)前事件循環(huán)
async def with_existing_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    await loop.run_until_complete(async_hello())

import asyncio
from asyncio import Future, Task
async def my_coro():
    """異步函數(shù)"""
    await asyncio.sleep(1)
    return "Done"
async def main():
    # 創(chuàng)建Future
    future = Future()
    # 在另一個(gè)協(xié)程中設(shè)置Future的值
    async def set_future():
        await asyncio.sleep(2)
        future.set_result("Future completed!")
    # 創(chuàng)建Task執(zhí)行協(xié)程
    asyncio.create_task(set_future())
    # 等待Future完成
    result = await future
    print(f"Future result: {result}")
    # 創(chuàng)建Task
    task = asyncio.create_task(my_coro())
    print(f"Task: {task}")
    print(f"Task done? {task.done()}")
    # 等待Task完成
    result = await task
    print(f"Task result: {result}")
asyncio.run(main())

import asyncio
import time
async def operation(name, duration):
    """模擬一個(gè)異步操作"""
    print(f"{name} 開始 (預(yù)計(jì) {duration}s)")
    start = time.time()
    await asyncio.sleep(duration)  # 暫停此協(xié)程
    elapsed = time.time() - start
    print(f"{name} 完成 (實(shí)際 {elapsed:.2f}s)")
    return f"{name} result"
async def sequential_execution():
    """順序執(zhí)行"""
    start = time.time()
    r1 = await operation("操作1", 2)
    r2 = await operation("操作2", 2)
    r3 = await operation("操作3", 2)
    total = time.time() - start
    print(f"順序執(zhí)行總耗時(shí): {total:.2f}s")
    return [r1, r2, r3]
async def concurrent_execution():
    """并發(fā)執(zhí)行"""
    start = time.time()
    # 創(chuàng)建三個(gè)任務(wù)并發(fā)執(zhí)行
    r1, r2, r3 = await asyncio.gather(
        operation("操作1", 2),
        operation("操作2", 2),
        operation("操作3", 2),
    )
    total = time.time() - start
    print(f"并發(fā)執(zhí)行總耗時(shí): {total:.2f}s")
    return [r1, r2, r3]
async def main():
    print("=" * 50)
    print("順序執(zhí)行:")
    await sequential_execution()
    print("=" * 50)
    print("并發(fā)執(zhí)行:")
    await concurrent_execution()
asyncio.run(main())
# 輸出：
# 順序執(zhí)行總耗時(shí): 6.02s（2+2+2）
# 并發(fā)執(zhí)行總耗時(shí): 2.01s（三個(gè)任務(wù)同時(shí)執(zhí)行）

import asyncio
class AsyncTimer:
    """異步計(jì)時(shí)器上下文管理器"""
    def __init__(self, name):
        self.name = name
        self.start_time = None
    async def __aenter__(self):
        """進(jìn)入上下文時(shí)執(zhí)行"""
        self.start_time = asyncio.get_event_loop().time()
        print(f"[{self.name}] 開始")
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出上下文時(shí)執(zhí)行"""
        elapsed = asyncio.get_event_loop().time() - self.start_time
        print(f"[{self.name}] 結(jié)束，耗時(shí): {elapsed:.2f}s")
        # 返回True可以抑制異常
        return False
async def main():
    async with AsyncTimer("任務(wù)1"):
        await asyncio.sleep(1)
    async with AsyncTimer("任務(wù)2"):
        await asyncio.sleep(0.5)
asyncio.run(main())

import asyncio
class AsyncCounter:
    """異步計(jì)數(shù)器迭代器"""
    def __init__(self, max_value):
        self.max_value = max_value
        self.current = 0
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.current >= self.max_value:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)  # 模擬異步操作
        value = self.current
        self.current += 1
        return value
async def main():
    async for i in AsyncCounter(5):
        print(f"計(jì)數(shù): {i}")
asyncio.run(main())
# 異步生成器
async def async_range(start, end, step=1):
    """異步生成器"""
    current = start
    while current < end:
        await asyncio.sleep(0.1)
        yield current
        current += step
async def main2():
    async for i in async_range(0, 5):
        print(f"異步生成器: {i}")
asyncio.run(main2())

import asyncio
import random
async def fetch_url(url):
    """模擬獲取URL"""
    delay = random.uniform(0.5, 2.0)
    await asyncio.sleep(delay)
    return f"{url} - OK"
async def main():
    urls = [
        "https://example.com/api/users",
        "https://example.com/api/posts",
        "https://example.com/api/comments",
        "https://example.com/api/tags",
    ]
    # 并發(fā)執(zhí)行所有URL請(qǐng)求
    results = await asyncio.gather(*[fetch_url(url) for url in urls])
    for url, result in zip(urls, results):
        print(f"{url}: {result}")
asyncio.run(main())

import asyncio
async def task_with_error(task_id):
    """可能出錯(cuò)的任務(wù)"""
    await asyncio.sleep(task_id * 0.5)
    if task_id == 2:
        raise ValueError(f"Task {task_id} failed!")
    return f"Task {task_id} completed"
async def main():
    # 使用TaskGroup
    async with asyncio.TaskGroup() as tg:
        # 創(chuàng)建任務(wù)
        tasks = []
        for i in range(4):
            task = tg.create_task(task_with_error(i))
            tasks.append(task)
        # TaskGroup會(huì)自動(dòng)等待所有任務(wù)完成
        # 如果任一任務(wù)拋出異常，其他任務(wù)會(huì)被取消
    # 訪問結(jié)果
    for task in tasks:
        if task.result() is not None:
            print(f"Result: {task.result()}")
asyncio.run(main())

import asyncio
import time
async def limited_task(task_id, semaphore):
    """限流任務(wù)"""
    async with semaphore:  # 獲取信號(hào)量
        print(f"Task {task_id} 開始")
        await asyncio.sleep(1)
        print(f"Task {task_id} 結(jié)束")
        return f"Task {task_id} done"
async def main():
    # 創(chuàng)建信號(hào)量，限制最多3個(gè)并發(fā)
    semaphore = asyncio.Semaphore(3)
    start = time.time()
    # 創(chuàng)建10個(gè)任務(wù)，但最多只有3個(gè)同時(shí)執(zhí)行
    tasks = [limited_task(i, semaphore) for i in range(10)]
    results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    print(f"總耗時(shí): {elapsed:.2f}s")  # 約4秒（10/3 ≈ 4）
    print(f"結(jié)果: {results}")
asyncio.run(main())

import asyncio
import random
async def producer(queue, producer_id):
    """生產(chǎn)者"""
    for i in range(3):
        item = f"P{producer_id}-Item{i}"
        await queue.put(item)
        print(f"生產(chǎn)者 {producer_id} 生產(chǎn): {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))
async def consumer(queue, consumer_id):
    """消費(fèi)者"""
    while True:
        try:
            item = await asyncio.wait_for(queue.get(), timeout=2.0)
            print(f"消費(fèi)者 {consumer_id} 消費(fèi): {item}")
            await asyncio.sleep(1)  # 模擬處理
            queue.task_done()
        except asyncio.TimeoutError:
            print(f"消費(fèi)者 {consumer_id} 等待超時(shí)，退出")
            break
async def main():
    queue = asyncio.Queue(maxsize=10)
    # 創(chuàng)建生產(chǎn)者和消費(fèi)者
    producers = [asyncio.create_task(producer(queue, i)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]
    # 等待所有生產(chǎn)者完成
    await asyncio.gather(*producers)
    # 等待隊(duì)列清空
    await queue.join()
    # 取消所有消費(fèi)者
    for c in consumers:
        c.cancel()
    await asyncio.gather(*consumers, return_exceptions=True)
asyncio.run(main())

import asyncio
import aiohttp
async def fetch_with_aiohttp(url, session):
    """使用aiohttp獲取URL"""
    async with session.get(url) as response:
        if response.status == 200:
            return await response.text()
        return None
async def concurrent_requests(urls):
    """并發(fā)請(qǐng)求多個(gè)URL"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_aiohttp(url, session) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
async def fetch_with_headers():
    """帶請(qǐng)求頭的請(qǐng)求"""
    async with aiohttp.ClientSession() as session:
        headers = {
            "User-Agent": "AsyncBot/1.0",
            "Accept": "application/json",
        }
        async with session.get(
            "https://httpbin.org/headers",
            headers=headers
        ) as response:
            if response.status == 200:
                return await response.json()
async def post_request():
    """POST請(qǐng)求"""
    async with aiohttp.ClientSession() as session:
        payload = {"username": "test", "password": "secret"}
        async with session.post(
            "https://httpbin.org/post",
            json=payload
        ) as response:
            return await response.json()
async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
    ]
    print("并發(fā)請(qǐng)求:")
    start = time.time()
    results = await concurrent_requests(urls)
    print(f"耗時(shí): {time.time() - start:.2f}s")
    print(f"結(jié)果數(shù)量: {len(results)}")
    print("\n帶Headers請(qǐng)求:")
    headers_result = await fetch_with_headers()
    print(headers_result)
    print("\nPOST請(qǐng)求:")
    post_result = await post_request()
    print(post_result.get("json", {}))
import time
asyncio.run(main())

import asyncio
import aiohttp
async def fetch_with_pool():
    """使用連接池"""
    # 創(chuàng)建連接限制器
    connector = aiohttp.TCPConnector(limit=10, limit_per_host=5)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/get") as response:
            return await response.json()
async def fetch_with_timeout():
    """設(shè)置超時(shí)"""
    timeout = aiohttp.ClientTimeout(total=30, connect=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get("https://httpbin.org/delay/5") as response:
                return await response.text()
        except asyncio.TimeoutError:
            print("請(qǐng)求超時(shí)!")
            return None
async def retry_request(url, max_retries=3):
    """帶重試的請(qǐng)求"""
    for attempt in range(max_retries):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    if response.status < 500:
                        return await response.json()
                    raise aiohttp.ClientError(f"HTTP {response.status}")
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            print(f"嘗試 {attempt + 1} 失敗: {e}")
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 指數(shù)退避
async def main():
    result = await fetch_with_pool()
    print("連接池請(qǐng)求完成:", result.get("url"))
    result = await fetch_with_timeout()
    print("超時(shí)請(qǐng)求:", result)
asyncio.run(main())

from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from pydantic import BaseModel, EmailStr
from typing import Optional, List
import asyncio
import time
app = FastAPI(title="異步API示例", version="1.0.0")
# Pydantic模型定義請(qǐng)求和響應(yīng)
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str
class User(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool = True
class OrderItem(BaseModel):
    product_id: int
    quantity: int
    price: float
class OrderCreate(BaseModel):
    user_id: int
    items: List[OrderItem]
# 模擬數(shù)據(jù)庫
users_db = []
orders_db = []
# 異步依賴注入
async def get_db():
    """模擬數(shù)據(jù)庫會(huì)話"""
    await asyncio.sleep(0.1)  # 模擬數(shù)據(jù)庫延遲
    return {"users": users_db, "orders": orders_db}
@app.get("/")
async def root():
    return {"message": "FastAPI 異步API示例"}
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    """獲取用戶"""
    await asyncio.sleep(0.05)  # 模擬DB查詢
    for user in users_db:
        if user["id"] == user_id:
            return User(**user)
    raise HTTPException(status_code=404, detail="用戶不存在")
@app.post("/users/", response_model=User, status_code=201)
async def create_user(user: UserCreate):
    """創(chuàng)建用戶"""
    # 檢查用戶名是否已存在
    if any(u["username"] == user.username for u in users_db):
        raise HTTPException(status_code=400, detail="用戶名已存在")
    new_user = {
        "id": len(users_db) + 1,
        "username": user.username,
        "email": user.email,
        "is_active": True
    }
    users_db.append(new_user)
    return User(**new_user)
@app.get("/users/", response_model=List[User])
async def list_users(skip: int = 0, limit: int = 10):
    """獲取用戶列表（分頁）"""
    await asyncio.sleep(0.05)
    return [User(**u) for u in users_db[skip:skip+limit]]
@app.post("/orders/", status_code=201)
async def create_order(order: OrderCreate, db=Depends(get_db)):
    """創(chuàng)建訂單"""
    # 驗(yàn)證用戶存在
    user_exists = any(u["id"] == order.user_id for u in db["users"])
    if not user_exists:
        raise HTTPException(status_code=404, detail="用戶不存在")
    # 計(jì)算總金額
    total = sum(item.quantity * item.price for item in order.items)
    new_order = {
        "id": len(db["orders"]) + 1,
        "user_id": order.user_id,
        "items": [item.dict() for item in order.items],
        "total": total,
        "status": "pending"
    }
    db["orders"].append(new_order)
    return new_order
@app.get("/orders/{order_id}")
async def get_order(order_id: int, db=Depends(get_db)):
    """獲取訂單"""
    for order in db["orders"]:
        if order["id"] == order_id:
            return order
    raise HTTPException(status_code=404, detail="訂單不存在")
@app.delete("/orders/{order_id}")
async def cancel_order(order_id: int, db=Depends(get_db)):
    """取消訂單"""
    for order in db["orders"]:
        if order["id"] == order_id:
            if order["status"] != "pending":
                raise HTTPException(status_code=400, detail="只能取消待處理訂單")
            order["status"] = "cancelled"
            return {"message": "訂單已取消"}
    raise HTTPException(status_code=404, detail="訂單不存在")
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import asyncio
app = FastAPI()
class EmailRequest(BaseModel):
    to: str
    subject: str
    body: str
def send_email_sync(email: EmailRequest):
    """同步發(fā)送郵件（模擬）"""
    import time
    print(f"發(fā)送郵件到 {email.to}...")
    time.sleep(5)  # 模擬耗時(shí)操作
    print(f"郵件已發(fā)送: {email.subject}")
async def send_email_async(email: EmailRequest):
    """異步發(fā)送郵件"""
    print(f"異步發(fā)送郵件到 {email.to}...")
    await asyncio.sleep(5)  # 模擬I/O操作
    print(f"郵件已發(fā)送: {email.subject}")
@app.post("/send-email-sync")
def endpoint_sync(email: EmailRequest, background_tasks: BackgroundTasks):
    """同步郵件發(fā)送（后臺(tái)執(zhí)行）"""
    background_tasks.add_task(send_email_sync, email)
    return {"message": "郵件發(fā)送任務(wù)已添加"}
@app.post("/send-email-async")
async def endpoint_async(email: EmailRequest):
    """異步郵件發(fā)送"""
    await send_email_async(email)
    return {"message": "郵件已發(fā)送"}
@app.post("/send-email-bg")
async def endpoint_bg(email: EmailRequest, background_tasks: BackgroundTasks):
    """混合模式：后臺(tái)任務(wù)中的異步函數(shù)"""
    background_tasks.add_task(send_email_async, email)
    return {"message": "郵件發(fā)送任務(wù)已在后臺(tái)添加"}

import asyncio
import time
# 錯(cuò)誤：在異步代碼中使用time.sleep
async def bad_example():
    print("開始")
    time.sleep(5)  # 阻塞整個(gè)事件循環(huán)！
    print("結(jié)束")
# 正確：使用asyncio.sleep
async def good_example():
    print("開始")
    await asyncio.sleep(5)  # 只暫停當(dāng)前協(xié)程
    print("結(jié)束")
# 對(duì)于必須使用阻塞I/O的場(chǎng)景，使用線程池
async def run_blocking_in_executor():
    """使用線程池執(zhí)行阻塞代碼"""
    def blocking_io():
        import time
        time.sleep(3)
        return "Blocking result"
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_io)
    return result

import asyncio
async def may_fail():
    """可能失敗的任務(wù)"""
    await asyncio.sleep(1)
    if True:  # 模擬失敗條件
        raise ValueError("Something went wrong!")
    return "Success"
async def main():
    # 方法1：try-except
    try:
        result = await may_fail()
        print(f"結(jié)果: {result}")
    except ValueError as e:
        print(f"捕獲異常: {e}")
    # 方法2：asyncio.gather with return_exceptions
    results = await asyncio.gather(
        asyncio.sleep(1, result=1),
        may_fail(),
        asyncio.sleep(1, result=3),
        return_exceptions=True
    )
    print(f"gather結(jié)果: {results}")  # [1, ValueError(...), 3]
asyncio.run(main())

import asyncio
async def long_running_task(task_id):
    """長(zhǎng)時(shí)間運(yùn)行的任務(wù)"""
    try:
        for i in range(10):
            print(f"任務(wù) {task_id}: 第 {i} 步")
            await asyncio.sleep(1)
        return f"任務(wù) {task_id} 完成"
    except asyncio.CancelledError:
        print(f"任務(wù) {task_id} 被取消")
        raise
async def main():
    task = asyncio.create_task(long_running_task(1))
    # 等待5秒后取消任務(wù)
    await asyncio.sleep(5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("任務(wù)已被成功取消")
asyncio.run(main())

import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Optional
import re
@dataclass
class Article:
    """文章數(shù)據(jù)結(jié)構(gòu)"""
    title: str
    url: str
    author: Optional[str]
    publish_date: Optional[str]
    content: Optional[str]
class AsyncCrawler:
    """異步爬蟲"""
    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session: Optional[aiohttp.ClientSession] = None
        self.results: List[Article] = []
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    async def fetch_page(self, url: str) -> Optional[str]:
        """獲取單個(gè)頁面"""
        async with self.semaphore:
            try:
                async with self.session.get(url, timeout=30) as response:
                    if response.status == 200:
                        return await response.text()
            except Exception as e:
                print(f"獲取失敗 {url}: {e}")
            return None
    async def parse_article(self, url: str) -> Optional[Article]:
        """解析文章"""
        html = await self.fetch_page(url)
        if not html:
            return None
        # 簡(jiǎn)化解析邏輯
        title_match = re.search(r'<h1[^>]*>(.*?)</h1>', html, re.DOTALL)
        title = title_match.group(1).strip() if title_match else "Unknown"
        return Article(
            title=title,
            url=url,
            author=None,
            publish_date=None,
            content=None
        )
    async def crawl(self, urls: List[str]) -> List[Article]:
        """爬取多個(gè)URL"""
        tasks = [self.parse_article(url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for result in results:
            if isinstance(result, Article):
                self.results.append(result)
        return self.results
async def main():
    urls = [
        "https://example.com/article1",
        "https://example.com/article2",
        "https://example.com/article3",
    ]
    async with AsyncCrawler(max_concurrent=3) as crawler:
        articles = await crawler.crawl(urls)
        print(f"成功爬取 {len(articles)} 篇文章")
asyncio.run(main())