Chapter 12 · Concurrent Requests and Concurrency Control

Chapter goals: Master rate limiting, flow control, and concurrency control for concurrent requests, writing robust high-concurrency HTTP clients.

12.1 Concurrency Is Not Always Better

Unlimited concurrency can cause:

• Target server overload (possible IP ban)
• Local resource exhaustion (file descriptors, memory)
• Triggering rate limits

12.2 Semaphore Concurrency Control

import asyncio
import httpx

async def fetch_with_semaphore(urls: list[str], max_concurrent: int = 5):
    """Limit maximum concurrent requests"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def limited_fetch(client: httpx.AsyncClient, url: str):
        async with semaphore:  # Control concurrency
            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 Exponential Backoff Retry

import asyncio
import random

async def retry_with_backoff(client: httpx.AsyncClient, url: str, max_retries: int = 3):
    """Retry with exponential backoff"""
    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:  # Rate limited
                wait = min(2 ** attempt + random.random(), 60)
                print(f'429 Rate limited, waiting {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 Batch Request Pattern

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]:
    """Generic batch request framework"""
    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)

# Usage example
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 Chapter Summary

• Use Semaphore to limit concurrency, avoiding server overload;
• 429 rate limiting should implement exponential backoff to avoid avalanche;
• Batch request frameworks are reusable, improving code quality.

🧪 随堂测验

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

1. What is the main purpose of asyncio.Semaphore in HTTP concurrency?

2. What are the pros and cons of exponential backoff retry?

3. What does asyncio.gather(*tasks, return_exceptions=True) do?

4. Which is NOT a best practice for concurrent HTTP requests?

🛠️ Hands-on Practice

1. Implement a concurrent crawler with Semaphore, limiting 10 concurrent requests to scrape pages.
2. Implement exponential backoff retry, simulating 429 responses to test wait logic.
3. Encapsulate a generic batch_request function supporting custom concurrency and error handling.

Next chapter: HTTP/2 and WebSocket