掌握异步HTTP请求的高效处理艺术

在Python异步编程中，网络请求是最常见的I/O密集型任务之一。协程通过非阻塞方式处理网络请求，可以大幅提升程序的并发性能。本文将深入讲解如何使用协程高效处理网络请求，涵盖从基础到高级的完整解决方案。

一、协程网络请求基础

1.1 为什么使用协程处理网络请求？

• 高并发：单线程可处理数千并发连接
• 低资源：比线程更轻量（KB级内存占用）
• 高性能：避免同步阻塞，充分利用等待时间
• 代码简洁：async/await语法更直观

1.2 核心工具对比

工具	类型	特点	适用场景
aiohttp	异步HTTP库	完整HTTP客户端/服务器实现	通用HTTP请求
httpx	异步/同步库	兼容同步接口，功能丰富	需要同步/异步切换
requests	同步库	简单易用，生态丰富	简单脚本，非协程环境

---

二、aiohttp核心用法

2.1 安装与基础请求

# 安装
pip install aiohttp

# 基本GET请求
import aiohttp  # 导入异步HTTP客户端库
import asyncio   # 导入异步IO库

# 定义异步获取URL内容的函数
async def fetch(url):
    """
    使用aiohttp发送GET请求并返回响应内容
    
    参数:
        url (str): 要请求的URL
        
    返回:
        str: 响应的文本内容
    """
    # 创建客户端会话（自动管理连接池）
    async with aiohttp.ClientSession() as session:
        # 发送GET请求
        async with session.get(url) as response:
            # 读取并返回响应内容
            return await response.text()

# 主异步函数
async def main():
    """
    主函数，协调异步任务
    """
    # 调用fetch函数获取网页内容
    html = await fetch('https://example.com')
    
    # 打印前200个字符（避免输出过长）
    print(html[:200])  # 打印前200字符

# 程序入口
if __name__ == '__main__':
    # 创建事件循环并运行主函数
    asyncio.run(main())

2.2 请求参数设置

async def fetch_with_params():
    """
    带参数的GET请求
    
    语法结构:
        async with session.get(url, params=params, headers=headers, cookies=cookies)
    """
    async with aiohttp.ClientSession() as session:
        # 查询参数
        params = {'key1': 'value1', 'key2': 'value2'}
        # 请求头
        headers = {'User-Agent': 'MyApp/1.0'}
        # Cookies
        cookies = {'session_id': '12345'}
        
        # 发送带参数的GET请求
        async with session.get(
            'https://api.example.com/data',
            params=params,   # 查询参数
            headers=headers, # 请求头
            cookies=cookies  # Cookies
        ) as response:
            # 返回JSON格式的响应
            return await response.json()

2.3 POST请求与数据提交

async def post_data():
    """
    POST请求示例，支持表单数据、JSON数据和文件上传
    
    语法结构:
        async with session.post(url, data=form_data, json=json_data)
    """
    async with aiohttp.ClientSession() as session:
        # 表单数据提交
        form_data = aiohttp.FormData()
        form_data.add_field('username', 'admin')
        form_data.add_field('password', 'secret')
        
        # JSON数据提交
        json_data = {'title': 'Post Title', 'body': 'Content'}
        
        # 文件上传
        with open('image.jpg', 'rb') as f:
            files = {'file': f}
            
            # 发送POST请求
            async with session.post(
                'https://api.example.com/submit',
                data=form_data,  # 表单数据
                # json=json_data, # JSON数据
                # files=files     # 文件上传
            ) as response:
                return await response.text()

---

三、高效并发处理

3.1 基本并发模式

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.create_task(fetch_url(session, url))
            tasks.append(task)
        
        return await asyncio.gather(*tasks)

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    results = await fetch_all(urls)
    for url, content in zip(urls, results):
        print(f"{url}: {len(content)} bytes")

asyncio.run(main())

3.2 使用信号量控制并发量

async def bounded_fetch(session, url, semaphore):
    async with semaphore:
        return await fetch_url(session, url)

async def fetch_all(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks)

3.3 流式处理大响应

async def stream_large_response(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            with open('large_file.zip', 'wb') as f:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f.write(chunk)
                    print(f"已下载 {f.tell()} 字节")

---

四、高级技巧与最佳实践

4.1 连接池配置

from aiohttp import TCPConnector

async def custom_session():
    connector = TCPConnector(
        limit=100,              # 最大连接数
        limit_per_host=20,      # 单主机最大连接
        ssl=False,              # 禁用SSL验证（仅测试用）
        force_close=True        # 强制关闭空闲连接
    )
    
    timeout = aiohttp.ClientTimeout(
        total=60,               # 总超时
        connect=10,              # 连接超时
        sock_read=30             # 读取超时
    )
    
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={'User-Agent': 'MyApp/1.0'}
    ) as session:
        # 使用自定义session
        pass

4.2 重试机制

async def fetch_with_retry(session, url, retries=3, backoff=1):
    for attempt in range(retries):
        try:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.text()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < retries - 1:
                wait = backoff * (2 ** attempt)
                print(f"请求失败: {e}, {wait}秒后重试...")
                await asyncio.sleep(wait)
            else:
                raise

4.3 超时控制

async def fetch_with_timeout(session, url, timeout=10):
    try:
        async with session.get(url, timeout=timeout) as response:
            return await response.text()
    except asyncio.TimeoutError:
        print(f"请求超时: {url}")
        return None

4.4 速率限制

from datetime import datetime
import time

class RateLimiter:
    def __init__(self, calls_per_second):
        self.period = 1.0 / calls_per_second
        self.last_call = 0
    
    async def wait(self):
        now = time.time()
        elapsed = now - self.last_call
        if elapsed < self.period:
            await asyncio.sleep(self.period - elapsed)
        self.last_call = time.time()

async def rate_limited_fetch(session, url, limiter):
    await limiter.wait()
    return await fetch_url(session, url)

---

五、错误处理与监控

5.1 统一错误处理

async def safe_fetch(session, url):
    try:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.text()
    except aiohttp.ClientError as e:
        print(f"客户端错误: {e}")
    except asyncio.TimeoutError:
        print(f"请求超时: {url}")
    except Exception as e:
        print(f"未知错误: {e}")
    return None

5.2 性能监控

async def monitored_fetch(session, url):
    start = time.monotonic()
    try:
        async with session.get(url) as response:
            content = await response.text()
            duration = time.monotonic() - start
            print(f"{url} 耗时: {duration:.2f}s, 状态: {response.status}")
            return content
    except Exception as e:
        duration = time.monotonic() - start
        print(f"{url} 失败: {duration:.2f}s, 错误: {str(e)}")
        raise

5.3 日志记录

import logging
logging.basicConfig(level=logging.INFO)

async def logged_fetch(session, url):
    try:
        logging.info(f"开始请求: {url}")
        async with session.get(url) as response:
            content = await response.text()
            logging.info(f"请求完成: {url}, 状态: {response.status}")
            return content
    except Exception as e:
        logging.error(f"请求失败: {url}, 错误: {str(e)}")
        return None

---

六、实战案例：并发API请求

6.1 天气API并发查询

import asyncio
import aiohttp
from datetime import datetime

CITIES = ['London', 'Paris', 'Tokyo', 'New York', 'Sydney']
API_KEY = 'your_api_key'
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather'

async def get_weather(session, city):
    params = {
        'q': city,
        'appid': API_KEY,
        'units': 'metric'
    }
    async with session.get(BASE_URL, params=params) as response:
        data = await response.json()
        return {
            'city': city,
            'temp': data['main']['temp'],
            'humidity': data['main']['humidity'],
            'description': data['weather'][0]['description']
        }

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [get_weather(session, city) for city in CITIES]
        results = await asyncio.gather(*tasks)
        
        print(f"{'城市':<10}{'温度(℃)':<10}{'湿度(%)':<10}{'天气状况':<15}")
        for weather in results:
            print(f"{weather['city']:<10}{weather['temp']:<10}{weather['humidity']:<10}{weather['description']:<15}")

if __name__ == '__main__':
    start = datetime.now()
    asyncio.run(main())
    print(f"总耗时: {(datetime.now() - start).total_seconds():.2f}秒")

6.2 网站状态监控

import asyncio
import aiohttp

WEBSITES = [
    'https://google.com',
    'https://github.com',
    'https://python.org',
    'https://amazon.com',
    'https://netflix.com'
]

async def check_site(session, url):
    try:
        start = asyncio.get_event_loop().time()
        async with session.get(url, timeout=10) as response:
            elapsed = asyncio.get_event_loop().time() - start
            return {
                'url': url,
                'status': response.status,
                'latency': f"{elapsed:.3f}s"
            }
    except Exception as e:
        return {
            'url': url,
            'status': 'ERROR',
            'error': str(e)
        }

async def monitor_sites():
    async with aiohttp.ClientSession() as session:
        while True:
            tasks = [check_site(session, url) for url in WEBSITES]
            results = await asyncio.gather(*tasks)
            
            print("\n网站状态监控:")
            for result in results:
                if 'status' in result and result['status'] == 200:
                    print(f"{result['url']} - 正常 (延迟: {result['latency']})")
                else:
                    print(f"{result['url']} - 故障: {result.get('error', '未知错误')}")
            
            await asyncio.sleep(60)  # 每分钟检查一次

asyncio.run(monitor_sites())

---

七、性能优化技巧

7.1 连接复用

# 全局Session（适用于长期运行应用）
session = aiohttp.ClientSession()

async def fetch(url):
    async with session.get(url) as response:
        return await response.text()

# 应用结束时关闭
async def shutdown():
    await session.close()

7.2 DNS缓存

from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=["8.8.8.8", "1.1.1.1"])
connector = aiohttp.TCPConnector(resolver=resolver, use_dns_cache=True)

async with aiohttp.ClientSession(connector=connector) as session:
    # 使用带DNS缓存的session

7.3 响应压缩

async with session.get(url, compress=True) as response:
    # 服务器支持时会自动解压

7.4 连接保持

connector = aiohttp.TCPConnector(keepalive_timeout=30)

---

八、协程网络请求流程图

---

九、总结

协程处理网络请求的核心优势：

1. 高性能：非阻塞I/O实现高并发

2. 资源高效：单线程处理数千连接

3. 代码简洁：async/await语法直观

4. 灵活控制：精细管理超时、重试、并发

关键实践：

• 使用aiohttp或httpx作为HTTP客户端

• 通过ClientSession复用连接

• 使用信号量控制并发量

• 实现健壮的错误处理和重试机制

• 监控请求性能并优化超时设置

掌握这些技巧，您将能够构建出高性能、高可靠的网络请求处理系统，轻松应对高并发场景。