`Python`的`HTTP`客户端：`requests`和`httpx`的`异步`用法。

好的，开始今天的讲座。

讲座主题：Python HTTP 客户端：Requests 和 HTTPX 的异步用法

今天我们将深入探讨 Python 中两个流行的 HTTP 客户端库：requests 和 httpx，并重点关注它们的异步用法。异步编程在处理高并发和 I/O 密集型任务时至关重要，可以显著提高应用程序的性能和响应能力。

1. 为什么需要异步 HTTP 客户端？

传统的同步 HTTP 客户端在发起请求后会阻塞程序的执行，直到收到响应。在高并发场景下，这种阻塞会导致大量的线程或进程等待，从而消耗大量的系统资源，降低程序的整体性能。

异步 HTTP 客户端则采用非阻塞的方式发起请求，允许程序在等待响应期间继续执行其他任务。当响应到达时，会通过回调、future 或 async/await 机制通知程序进行处理。这样可以避免线程或进程的阻塞，提高资源利用率，并显著提升程序的并发性能。

2. Requests 库的异步支持 (AIOHTTP)

requests 库本身是同步的，但可以通过与 aiohttp 库结合来实现异步 HTTP 请求。aiohttp 是一个基于 asyncio 的异步 HTTP 客户端/服务端框架。

2.1 安装 aiohttp

首先，我们需要安装 aiohttp 库：

pip install aiohttp

2.2 异步 GET 请求

以下是一个使用 aiohttp 进行异步 GET 请求的示例：

import asyncio
import aiohttp

async def fetch_url(url: str, session: aiohttp.ClientSession) -> str:
    """
    异步获取指定 URL 的内容。

    Args:
        url: 要获取的 URL。
        session: aiohttp 客户端会话。

    Returns:
        URL 的内容。
    """
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 HTTP 请求。
    """
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://www.google.com"
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(url, session) for url in urls]
        results = await asyncio.gather(*tasks)

    for url, result in zip(urls, results):
        if result:
            print(f"Content from {url}: {result[:100]}...") # 打印前100个字符

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

代码解释：

• aiohttp.ClientSession 是 aiohttp 中用于管理客户端连接的类。它类似于 requests.Session，但用于异步操作。
• async with session.get(url) as response: 使用异步上下文管理器发起 GET 请求。await response.text() 异步地读取响应内容。
• asyncio.gather(*tasks) 并发地执行多个异步任务。它接收一个包含多个 async 函数的列表，并返回一个包含所有任务结果的列表。
• asyncio.run(main()) 是运行异步函数的标准方式。

2.3 异步 POST 请求

import asyncio
import aiohttp
import json

async def post_data(url: str, data: dict, session: aiohttp.ClientSession) -> dict:
    """
    异步发送 POST 请求。

    Args:
        url: 要发送请求的 URL。
        data: 要发送的 JSON 数据。
        session: aiohttp 客户端会话。

    Returns:
        响应的 JSON 数据。
    """
    try:
        async with session.post(url, json=data) as response:
            return await response.json()
    except Exception as e:
        print(f"Error posting data to {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 POST 请求。
    """
    url = "https://httpbin.org/post"  # 一个接收 POST 请求并返回数据的 API
    data = {"key1": "value1", "key2": "value2"}

    async with aiohttp.ClientSession() as session:
        result = await post_data(url, data, session)

    if result:
        print(f"Response from {url}: {result}")

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

代码解释：

• session.post(url, json=data) 使用 json 参数发送 JSON 数据。 aiohttp 会自动设置 Content-Type 为 application/json。
• await response.json() 异步地解析响应的 JSON 数据。

2.4 处理连接池

aiohttp.ClientSession 内部维护一个连接池，用于复用 TCP 连接，提高性能。 建议在应用程序的生命周期内重用同一个 ClientSession 实例，而不是为每个请求创建一个新的实例。 在 main 函数中使用 async with 语句可以确保在会话结束后正确关闭连接。

2.5 错误处理

在 fetch_url 和 post_data 函数中，我们使用了 try...except 块来捕获异常，并打印错误信息。 可以根据实际需求进行更详细的错误处理，例如重试请求、记录日志等。

3. HTTPX 库的异步支持

httpx 是一个现代化的、功能丰富的 HTTP 客户端库，它原生支持异步操作，并且提供了与 requests 类似的 API，使得从 requests 迁移到 httpx 变得非常容易。

3.1 安装 HTTPX

首先，我们需要安装 httpx 库：

pip install httpx

3.2 异步 GET 请求

以下是一个使用 httpx 进行异步 GET 请求的示例：

import asyncio
import httpx

async def fetch_url(url: str, client: httpx.AsyncClient) -> str:
    """
    异步获取指定 URL 的内容。

    Args:
        url: 要获取的 URL。
        client: httpx 异步客户端。

    Returns:
        URL 的内容。
    """
    try:
        response = await client.get(url)
        response.raise_for_status()  # 检查 HTTP 状态码
        return response.text
    except httpx.HTTPError as e:
        print(f"HTTP error fetching {url}: {e}")
        return None
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 HTTP 请求。
    """
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://www.google.com"
    ]

    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(url, client) for url in urls]
        results = await asyncio.gather(*tasks)

    for url, result in zip(urls, results):
        if result:
            print(f"Content from {url}: {result[:100]}...")  # 打印前100个字符

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

代码解释：

• httpx.AsyncClient 是 httpx 中用于异步 HTTP 请求的客户端类。它类似于 requests.Session 和 aiohttp.ClientSession。
• response = await client.get(url) 使用异步方式发起 GET 请求。
• response.raise_for_status() 检查 HTTP 状态码，如果状态码表示错误（例如 404、500），则抛出 httpx.HTTPError 异常。 这是一个良好的实践，可以确保程序能够及时发现并处理 HTTP 错误。
• response.text 异步地读取响应内容。

3.3 异步 POST 请求

import asyncio
import httpx
import json

async def post_data(url: str, data: dict, client: httpx.AsyncClient) -> dict:
    """
    异步发送 POST 请求。

    Args:
        url: 要发送请求的 URL。
        data: 要发送的 JSON 数据。
        client: httpx 异步客户端。

    Returns:
        响应的 JSON 数据。
    """
    try:
        response = await client.post(url, json=data)
        response.raise_for_status()
        return response.json()
    except httpx.HTTPError as e:
        print(f"HTTP error posting data to {url}: {e}")
        return None
    except Exception as e:
        print(f"Error posting data to {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 POST 请求。
    """
    url = "https://httpbin.org/post"  # 一个接收 POST 请求并返回数据的 API
    data = {"key1": "value1", "key2": "value2"}

    async with httpx.AsyncClient() as client:
        result = await post_data(url, data, client)

    if result:
        print(f"Response from {url}: {result}")

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

代码解释：

• client.post(url, json=data) 使用 json 参数发送 JSON 数据。httpx 会自动设置 Content-Type 为 application/json。
• response.json() 异步地解析响应的 JSON 数据。

3.4 超时设置

httpx 允许设置请求的超时时间，以防止程序长时间等待无响应的请求。

import asyncio
import httpx

async def fetch_url(url: str, client: httpx.AsyncClient) -> str:
    """
    异步获取指定 URL 的内容，并设置超时时间。

    Args:
        url: 要获取的 URL。
        client: httpx 异步客户端。

    Returns:
        URL 的内容。
    """
    try:
        response = await client.get(url, timeout=5.0)  # 设置超时时间为 5 秒
        response.raise_for_status()
        return response.text
    except httpx.TimeoutException:
        print(f"Timeout while fetching {url}")
        return None
    except httpx.HTTPError as e:
        print(f"HTTP error fetching {url}: {e}")
        return None
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 HTTP 请求。
    """
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://www.google.com",
        "https://www.example.com:81" #模拟一个无法访问的地址，测试超时
    ]

    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(url, client) for url in urls]
        results = await asyncio.gather(*tasks)

    for url, result in zip(urls, results):
        if result:
            print(f"Content from {url}: {result[:100]}...")
        else:
            print(f"Failed to retrieve content from {url}")

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

代码解释：

• client.get(url, timeout=5.0) 设置请求的超时时间为 5 秒。 如果在 5 秒内没有收到响应，httpx 将抛出 httpx.TimeoutException 异常。
• 在 try...except 块中捕获 httpx.TimeoutException 异常，并进行相应的处理。

3.5 代理设置

httpx 支持通过代理服务器发起 HTTP 请求。

import asyncio
import httpx

async def fetch_url(url: str, client: httpx.AsyncClient) -> str:
    """
    异步获取指定 URL 的内容，并通过代理服务器。

    Args:
        url: 要获取的 URL。
        client: httpx 异步客户端。

    Returns:
        URL 的内容。
    """
    try:
        response = await client.get(url)
        response.raise_for_status()
        return response.text
    except httpx.HTTPError as e:
        print(f"HTTP error fetching {url}: {e}")
        return None
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

async def main():
    """
    主函数，用于执行异步 HTTP 请求。
    """
    url = "https://www.example.com"
    proxies = {
        "http": "http://your_proxy_address:port",  # 替换为你的 HTTP 代理地址
        "https": "http://your_proxy_address:port", # 替换为你的 HTTPS 代理地址，如果不同的话
    }

    async with httpx.AsyncClient(proxies=proxies) as client:
        result = await fetch_url(url, client)

    if result:
        print(f"Content from {url}: {result[:100]}...")

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

代码解释：

• httpx.AsyncClient(proxies=proxies) 使用 proxies 参数设置代理服务器。 proxies 是一个字典，包含 http 和 https 两个键，分别对应 HTTP 和 HTTPS 代理地址。 需要将 your_proxy_address 和 port 替换为实际的代理地址和端口。

4. Requests (AIOHTTP) vs. HTTPX：对比分析

特性	Requests (AIOHTTP)	HTTPX
异步支持	通过 aiohttp 库实现	原生支持
API 相似性	与 requests 相似	与 requests 更相似，迁移成本更低
性能	优秀，aiohttp 基于 asyncio	优秀，基于 asyncio 和 trio
HTTP/2 支持	支持	支持
类型提示	部分支持	良好支持，有助于代码维护
文档	完善	完善
依赖	依赖 aiohttp	无强制依赖，但可以与 trio 集成
同步支持	同步请求使用 requests，异步请求使用 aiohttp	同一个库同时支持同步和异步请求，简化代码结构

选择建议：

• 如果已经在使用 requests 并且只需要添加异步支持，aiohttp 是一个不错的选择。
• 如果需要一个现代化的、功能丰富的、原生支持异步的 HTTP 客户端库，httpx 是更好的选择。
• 如果需要同时支持同步和异步请求，并且希望使用同一个库，httpx 是唯一的选择。

5. 最佳实践

• 重用客户端会话： 尽量在应用程序的生命周期内重用同一个 aiohttp.ClientSession 或 httpx.AsyncClient 实例，而不是为每个请求创建一个新的实例。
• 设置超时时间： 为请求设置合理的超时时间，以防止程序长时间等待无响应的请求。
• 处理异常： 捕获并处理 HTTP 错误和其他异常，以确保程序的健壮性。
• 使用连接池： 利用 aiohttp 和 httpx 提供的连接池功能，提高性能。
• 异步上下文管理器： 使用 async with 语句管理客户端会话，确保在会话结束后正确关闭连接。
• 合理使用并发： 不要过度使用并发，避免对服务器造成过大的压力。
• 监控和日志： 监控 HTTP 请求的性能指标，并记录日志，以便进行问题排查和性能优化。

6. 案例分析：异步爬虫

异步 HTTP 客户端非常适合用于构建高性能的 Web 爬虫。以下是一个使用 httpx 构建异步爬虫的简单示例：

import asyncio
import httpx
import re

async def fetch_url(url: str, client: httpx.AsyncClient, visited_urls: set) -> None:
    """
    异步获取指定 URL 的内容，并提取其中的链接。

    Args:
        url: 要获取的 URL。
        client: httpx 异步客户端。
        visited_urls: 已经访问过的 URL 集合。
    """
    if url in visited_urls:
        return

    visited_urls.add(url)

    try:
        response = await client.get(url)
        response.raise_for_status()
        html = response.text
        print(f"Crawled: {url}")

        # 提取链接
        links = re.findall(r'<a href="(.*?)"', html)
        for link in links:
            # 简单的URL规范化，仅供示例
            if link.startswith("/"):
                next_url = url + link
            elif link.startswith("http"):
                next_url = link
            else:
                continue #忽略相对链接

            asyncio.create_task(fetch_url(next_url, client, visited_urls)) #异步创建新任务

    except httpx.HTTPError as e:
        print(f"HTTP error fetching {url}: {e}")
    except Exception as e:
        print(f"Error fetching {url}: {e}")

async def main():
    """
    主函数，用于启动异步爬虫。
    """
    start_url = "https://www.example.com" # 替换为你想要爬取的起始 URL
    visited_urls = set()

    async with httpx.AsyncClient() as client:
        await fetch_url(start_url, client, visited_urls)

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

代码解释：

• fetch_url 函数异步地获取指定 URL 的内容，并使用正则表达式提取其中的链接。
• visited_urls 集合用于记录已经访问过的 URL，避免重复爬取。
• asyncio.create_task(fetch_url(next_url, client, visited_urls)) 异步地创建新的任务来爬取提取到的链接。 这使得爬虫可以并发地爬取多个页面。
• 需要注意的是，这个示例非常简化，仅用于演示异步爬虫的基本原理。 在实际应用中，需要考虑更多的因素，例如 robots.txt 协议、反爬虫机制、数据存储等。

7. 总结：异步HTTP客户端是提升性能的关键

我们学习了如何使用 requests (通过 aiohttp) 和 httpx 进行异步 HTTP 请求。异步编程可以显著提高应用程序的性能和响应能力，尤其是在处理高并发和 I/O 密集型任务时。 httpx 提供了更现代化的 API 和更好的类型提示，而 requests 结合 aiohttp 也是一个成熟的选择。 选择哪个库取决于你的具体需求和偏好。

今天的讲座就到这里，希望对大家有所帮助！