📚 编程进阶教程站

第 12 章 · 并发请求与并发控制 ​

本章目标：掌握并发请求的限速、限流、并发度控制，写出健壮的高并发 HTTP 客户端。

12.1 并发不是越快越好 ​

无限制并发会导致：

  • 目标服务器过载（可能被封 IP）
  • 本地资源耗尽（文件描述符、内存）
  • 触发速率限制

12.2 Semaphore 并发控制 ​

python
import asyncio
import httpx

async def fetch_with_semaphore(urls: list[str], max_concurrent: int = 5):
    """限制最大并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def limited_fetch(client: httpx.AsyncClient, url: str):
        async with semaphore:  # 控制并发度
            resp = await client.get(url)
            return resp.status_code
    
    async with httpx.AsyncClient() as client:
        tasks = [limited_fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks)

12.3 指数退避重试 ​

python
import asyncio
import random

async def retry_with_backoff(client: httpx.AsyncClient, url: str, max_retries: int = 3):
    """带指数退避的重试"""
    for attempt in range(max_retries):
        try:
            resp = await client.get(url)
            resp.raise_for_status()
            return resp.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:  # 速率限制
                wait = min(2 ** attempt + random.random(), 60)
                print(f'429 限制，等待 {wait:.1f}s...')
                await asyncio.sleep(wait)
            elif e.response.status_code >= 500 and attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
            else:
                raise
    return None

12.4 批量请求模式 ​

python
import asyncio
import httpx
from typing import List, Callable, Any

async def batch_request(
    urls: List[str],
    handler: Callable[[httpx.AsyncClient, str], Any],
    concurrency: int = 10
) -> List[Any]:
    """通用批量请求框架"""
    results = []
    semaphore = asyncio.Semaphore(concurrency)
    
    async def wrapped(url: str):
        async with semaphore:
            async with httpx.AsyncClient() as client:
                return await handler(client, url)
    
    tasks = [wrapped(url) for url in urls]
    return await asyncio.gather(*tasks, return_exceptions=True)

# 使用示例
async def main():
    urls = ['https://api.github.com/users/' + u for u in ['octocat', 'torvalds']]
    
    async def get_user(client, url):
        resp = await client.get(url)
        return resp.json()['login']
    
    results = await batch_request(urls, get_user, concurrency=5)
    for user in results:
        print(user)

12.5 本章小结 ​

  • 使用 Semaphore 限制并发度，避免压垮服务端；
  • 429 速率限制应实现指数退避，避免雪崩；
  • 批量请求框架可复用，提高代码质量。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. asyncio.Semaphore 在 HTTP 并发中的主要用途是？

2. 指数退避重试的优缺点是？

3. asyncio.gather(*tasks, return_exceptions=True) 的作用是？

4. 并发 HTTP 请求的最佳实践不包括？

🛠️ 动手实践 ​

  1. 实现带 Semaphore 的并发爬虫，限制 10 个并发请求抓取页面。
  2. 实现指数退避重试，模拟 429 响应测试等待逻辑。
  3. 封装通用 batch_request 函数，支持自定义并发度和错误处理。

下一章：HTTP/2 与 WebSocket