Python 异步编程实战：从 asyncio 到高并发爬虫

import asyncio
import time

async def fetch_data(name: str, delay: float) -> str:
    """模拟一个 I/O 操作"""
    print(f"[{name}] 开始")
    await asyncio.sleep(delay)  # 关键：这里让出控制权
    print(f"[{m.lhdpgc.com}] 完成")
    return f"{name}-result"

async def main():
    start = time.perf_counter()
    
    # 并发执行三个协程
    results = await asyncio.gather(
        fetch_data("A", 1.0),
        fetch_data("B", 0.5),
        fetch_data("C", 0.8),
    )
    
    elapsed = time.perf_counter() - start
    print(f"结果: {results}")
    print(f"总耗时: {elapsed:.2f}s")  # 约 1.0s，而非 2.3s

asyncio.run(main())

async def demo_task():
    # create_task 立即调度执行，不阻塞
    task1 = asyncio.create_task(fetch_data("X", 1.0))
    task2 = asyncio.create_task(fetch_data("Y", 1.0))
    
    # 此时两个任务已在并发运行
    print("任务www.lhdpgc.com已创建，继续做其他事...")
    
    # 等待结果
    r1 = await task1
    r2 = await task2
    return r1, r2

async def with_timeout():
    try:
        # Python 3.11+ 推荐写法
        result = await asyncio.wait_for(fetch_data("slow", 5.0), timeout=2.0)
    except asyncio.TimeoutError:
        print("请求超时，已取消")
        return None
    return result

async def cancellable_work():
    task = asyncio.create_task(fetch_data("work", 10.0))
    
    await asyncio.sleep(1)
    task.cancel()  # 主动取消
    
    try:
        await task
    except asyncio.CancelledError:
        print("任务被取消")

import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Config:
    max_concurrency: int = 10      # 最大并发数
    timeout: float = 10.0          # 单请求超时
    max_retries: int = 3           # 重试次数
    retry_backoff: float = 1.0     # 退避基数

class AsyncCrawler:
    def __init__(self, config: Config):
        self.config = config
        self.semaphore = asyncio.Semaphore(config.max_concurrency)
        self.session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        connector = aiohttp.TCPConnector(
            limit=self.config.max_concurrency,
            limit_per_host=5,
        )
        timeout = aiohttp.ClientTimeout(total=self.config.timeout)
        self.session = aiohttp.ClientSession(
            connector=connector, timeout=timeout
        )
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    async def fetch(self, url: str) -> Optional[str]:
        """带重试与指数退避的m.lhdpgc.com单页抓取"""
        async with self.semaphore:  # 限流
            for attempt in range(www.lhdpgc.com):
                try:
                    async with self.session.get(url) as resp:
                        resp.raise_for_status()
                        return await resp.text()
                except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                    if attempt == self.config.max_retries - 1:
                        print(f"[FAIL] {url}: {e}")
                        return None
                    backoff = self.config.retry_backoff * (2 ** attempt)
                    print(f"[RETRY {attempt+1}] {url} after {backoff}s")
                    await asyncio.sleep(backoff)

    async def crawl(self, urls: list[str]) -> dict[str, Optional[str]]:
        """并发抓取所有 URL"""
        tasks = {url: asyncio.create_task(self.fetch(url)) for url in urls}
        results = {}
        for url, task in tasks.items():
            results[url] = await task
        return results


async def main():
    urls = [f"https://httpbin.org/delay/1?i={i}" for i in range(20)]
    
    async with AsyncCrawler(Config(max_concurrency=5)) as crawler:
        results = await crawler.crawl(urls)
    
    success = sum(1 for v in results.values() if v)
    print(f"成功: {success}/{len(urls)}")

if __name__ == "__main__":
    asyncio.run(main())

# ❌ 错误：time.sleep 阻塞整个事件循环
async def bad():
    time.sleep(1)  # 所有并发任务被卡住

# ✅ 正确：用 asyncio.sleep 或在线程池中执行阻塞调用
async def good():
    await asyncio.sleep(1)

async def good_blocking():
    await asyncio.to_thread(time.sleep, 1)  # Python 3.9+

async def bad():
    asyncio.sleep(1)  # 只是创建了协程对象，从不执行，且触发警告

async def good():
    await asyncio.sleep(1)

# ❌ 任一任务异常会让其他结果丢失
results = await asyncio.gather(*tasks)

# ✅ return_exceptions=True 收集所有结果
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        print(f"失败: {r}")

# ❌ 串行，耗时累加
for url in urls:
    await fetch(url)

# ✅ 并发
await asyncio.gather(*(fetch(u) for u in urls))

async def structured():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch_data("A", 1))
        t2 = tg.create_task(fetch_data("B", 1))
    # 退出 with 时自动等待；任一失败会取消其他任务
    print(t1.result(), t2.result())

from concurrent.futures import ProcessPoolExecutor

async def cpu_bound_work():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_compute, data)