Requests、HTTPX 与AIOHTTP：选择哪一个？

Python 提供许多 HTTP 客户端。如果您不熟悉 HTTP（超文本传输协议），它指的是整个网络的底层框架。

今天，我们将比较 Python 最受欢迎的三个 HTTP 客户端：Requests、HTTPX 和 AIOHTTP。如果您想了解其他可用的客户端，请查看此处。

简要概述

Requests 是 Python 的标准 HTTP 客户端，通过阻塞和同步操作来实现便捷使用。HTTPX 属于较新的异步客户端，其设计兼顾了速度和易用性。AIOHTTP 已经发布十多年。它是 Python 提供的首批且具有最好支持的异步 HTTP 客户端。

功能	Requests	HTTPX	AIOHTTP
结构	同步/阻塞	异步/非阻塞	异步/非阻塞
会话	是	是	是
并发性	否	是	是
HTTP/2 支持	否	是	是
性能	低	高	高
重试	自动	自动	手动
超时支持	根据请求	全面支持	全面支持
代理支持	是	是	是
易于使用	简单	困难	困难
用例	简单项目/原型设计	高性能	高性能

Python Requests

Python Requests 非常直观且易于使用。如果不需要极端前沿的性能，那么它是 Python 中发送 HTTP 请求的首选客户端。Python Requests 使用广泛、易于理解，并且是世界上资料最齐全的 Python HTTP 客户端。

使用以下命令即可安装此客户端。

安装

pip install requests

Requests 支持标准 HTTP 协议，甚至支持部分会话管理。通过会话，您可以创建与服务器的持久连接。与发出单个请求相比，这可以让您更快、更有效地检索数据。如果您只想发送基本（GET、POST、PUT 和 DELETE）请求并且不关心高性能，则 Requests 库可以满足您的所有 HTTP 需求。

您可以从网络上找到有关代理集成、用户代理以及各种其他内容的指南。

您可以参考下文的一些基本用法示例。

import requests

# make a simple request
response = requests.get("https://jsonplaceholder.typicode.com/posts")
print(response.status_code)

# use a session for multiple requests to the same server
with requests.Session() as client:
    for get_request in range(1000):
        response = client.get("https://jsonplaceholder.typicode.com/posts")
        print(response.status_code)

涉及异步操作时，Requests 就难以满足要求了。异步支持让您可以同时发出一批请求。然后，您可以等待所有这些请求。对于同步请求，您一次只能发送一个请求，并且必须等待服务器返回响应才能发出另一个请求。如果您的程序需要发送大量 HTTP 请求，使用 Requests 会给您的代码带来一些固有局限。

HTTPX

HTTPX 是这三个库中最新、最现代的一个，为我们提供了即刻可用的异步操作全面支持。虽然如此，它仍然具有用户友好且直观的句法。如果 Requests 无法满足要求，并且您需要提高性能而又不想要学习太多内容，可以使用 HTTPX。

借助 asyncio（异步输入和输出），您可以通过 await 关键字，编写充分利用异步响应的代码。这样，我们在等待响应时就可以继续操作，无需阻塞代码中的所有其他内容。通过这些异步操作，您可以发送大批量的请求。您可以发送 5 个、50 个甚至 100 个请求，不必再一次只发一个请求！

安装

pip install httpx

以下是一些 HTTPX 入门示例。

import httpx
import asyncio

# synchronous response
response = httpx.get("https://jsonplaceholder.typicode.com/posts")
print(response.status_code)

# basic async session usage
async def main():
    async with httpx.AsyncClient() as client:
        for get_request in range(1000):
            response = await client.get("https://jsonplaceholder.typicode.com/posts")
            print(response.status_code)
asyncio.run(main())

HTTPX 是编写新代码的理想选择，但它也有自己的局限性。其句法决定了我们很难用 HTTPX 遍历并替换现有的 Requests 代码库。编写异步代码需要一些样板，而除非您发送数千个请求，否则通常不值得花费额外的时间开发样板。

与 AIOHTTP（您很快就会了解到）相比，HTTPX 速度并不是很快。如果您想构建 Web 服务器或者复杂的网络，HTTPX 的生态系统还不成熟，不是理想的选择。HTTPX 最适合具有 HTTP/2 等现代功能的新型客户端应用程序。

AIOHTTP

在异步编程方面，AIOHTTP 早已在 Python 中广泛使用。无论您运行的是服务器、客户端应用程序还是分布式网络，AIOHTTP 都可以满足所有这些需求。然而，在这三个库（Requests、HTTPX 和 AIOHTTP）中，AIOHTTP 的学习曲线最陡。

我们只侧重了解简单的客户端用法，因此不会深入研究复杂的 AIOHTTP。与 HTTPX 一样，我们也可以使用 AIOHTTP 发送批量请求。但是，与 HTTPX 不同的是，AIOHTTP 不支持同步请求。不要一次性发送太多请求……否则会被服务器禁止。

安装

pip install aiohttp

查看下面的基本用法。

import aiohttp
import asyncio

# make a single request
async def main():
    async with aiohttp.ClientSession() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts")
        print(response)
        
asyncio.run(main())


# basic async session usage
async def main():
    with aiohttp.ClientSession() as client:
        for response in range(1000):
            response = await client.get("https://jsonplaceholder.typicode.com/posts")
            print(response)
            
asyncio.run(main())

AIOHTTP 是严格异步库。正如您在前文代码中所看到的，我们的单一请求示例所需的代码远多于头两个示例。无论我们需要发送多少个请求，都要设置一个异步会话，因此建议结合使用代理和 AIOHTTP。对于单一请求来说，这绝对绰绰有余了。

尽管 AIOHTTP 有很多优点，但它并不会完全取代 Python Requests，需要同时处理海量入站和出站请求的情况除外。此外，AIOHTTP 将大大提升性能。在构建复杂的应用程序并需要极其快速的通信时，可以使用 AIOHTTP。这个库最适合服务器、分布式网络和高度复杂的网络抓取应用程序。

性能对比

现在，我们将使用每个库构建小程序。要求很简单：打开一个客户端会话并执行 1000 个 API 请求。

首先，我们需要创建与服务器的会话。接着，我们执行 1000 个请求。我们将有两个数组：一个用于正确响应，一个用于错误响应。运行完成后，我们会打印出正确请求和错误请求的全部数量。如果我们收到了错误的请求，我们会将其状态代码打印到控制台。

在我们的异步示例（HTTPX 和 AIOHTTP）中，我们将使用 chunkify() 函数。此函数用于将数组分块。然后我们分批执行请求。例如，如果我们想要按每批 50 个的数量执行请求，我们将使用 chunkify() 创建批次，并使用 process_chunk() 一次性执行所有 50 个请求。

请查看以下函数。

def chunkify(iterable, size):
    iterator = iter(iterable)
    while chunk := list(islice(iterator, size)):
        yield chunk
        
        
async def process_chunk(client, urls, retries=3):
    tasks = [fetch(client, url, retries) for url in urls]
    return await asyncio.gather(*tasks)

Requests

这是我们使用 Requests 的代码。与我们稍后使用的两个异步示例相比，它非常简单。我们打开一个会话并使用 for 循环遍历请求。

import requests
import json
from datetime import datetime

start_time = datetime.now()
good_responses = []
bad_responses = []

with requests.Session() as client:

    for get_request in range(1000):
        response = client.get("https://jsonplaceholder.typicode.com/posts")
        status_code = response.status_code
        if status_code  == 200:
            good_responses.append(status_code)
        else:
            bad_responses.append(status_code)

end_time = datetime.now()

print("----------------Requests------------------")
print(f"Time elapsed: {end_time - start_time}")
print(f"Good Responses: {len(good_responses)}")
print(f"Bad Responses: {len(bad_responses)}")

for status_code in set(bad_responses):
    print(status_code)

Requests 仅用了 51 秒多就为我们完成了这项任务。这相当于每秒处理大约 20 个请求。如果不使用会话，发送每个请求预计需要 2 秒。这说明 Requests 的性能相当出色。

HTTPX

以下是 HTTPX 的代码。如前文所述，我们使用了 chunkify() 和 process_chunk() 两个函数。

import httpx
import asyncio
from datetime import datetime
from itertools import islice

def chunkify(iterable, size):
    iterator = iter(iterable)
    while chunk := list(islice(iterator, size)):
        yield chunk

async def fetch(client, url, retries=3):
    """Fetch a URL with retries."""
    for attempt in range(retries):
        try:
            response = await client.get(url)
            return response.status_code
        except httpx.RequestError as e:
            if attempt < retries - 1:
                await asyncio.sleep(1)
            else:
                return f"Error: {e}"

async def process_chunk(client, urls, retries=3):
    tasks = [fetch(client, url, retries) for url in urls]
    return await asyncio.gather(*tasks)

async def main():
    url = "https://jsonplaceholder.typicode.com/posts"
    total_requests = 1000
    chunk_size = 50

    good_responses = []
    bad_responses = []

    async with httpx.AsyncClient(timeout=10) as client:
        start_time = datetime.now()

        urls = [url] * total_requests
        for chunk in chunkify(urls, chunk_size):
            results = await process_chunk(client, chunk)
            for status in results:
                if isinstance(status, int) and status == 200:
                    good_responses.append(status)
                else:
                    bad_responses.append(status)

        end_time = datetime.now()

        print("----------------HTTPX------------------")
        print(f"Time elapsed: {end_time - start_time}")
        print(f"Good Responses: {len(good_responses)}")
        print(f"Bad Responses: {len(bad_responses)}")

        if bad_responses:
            print("Bad Status Codes or Errors:")
            for error in set(bad_responses):
                print(error)

asyncio.run(main())

以下是我们使用 HTTPX 时的输出。与 Requests 相比，其表现令人惊叹。所用的总时间刚刚超过 7 秒。这相当于每秒处理 139.47 个请求。HTTPX 的性能大约是 Requests 的 7 倍。

AIOHTTP

现在，我们将使用 AIOHTTP 执行相同的任务。我们采用与 HTTPX 示例相同的基本结构。这里唯一的主要区别是 AIOHTTP 客户端取代了 HTTPX 客户端。

import aiohttp
import asyncio
from datetime import datetime
from itertools import islice

def chunkify(iterable, size):
    iterator = iter(iterable)
    while chunk := list(islice(iterator, size)):
        yield chunk

async def fetch(session, url, retries=3):
    for attempt in range(retries):
        try:
            async with session.get(url) as response:
                return response.status
        except aiohttp.ClientError as e:
            if attempt < retries - 1:
                await asyncio.sleep(1)
            else:
                return f"Error: {e}"

async def process_chunk(session, urls):
    tasks = [fetch(session, url) for url in urls]
    return await asyncio.gather(*tasks)

async def main():
    url = "https://jsonplaceholder.typicode.com/posts"
    total_requests = 1000
    chunk_size = 50

    good_responses = []
    bad_responses = []

    async with aiohttp.ClientSession() as session:
        start_time = datetime.now()

        urls = [url] * total_requests
        for chunk in chunkify(urls, chunk_size):
            results = await process_chunk(session, chunk)
            for status in results:
                if isinstance(status, int) and status == 200:
                    good_responses.append(status)
                else:
                    bad_responses.append(status)

        end_time = datetime.now()

        print("----------------AIOHTTP------------------")
        print(f"Time elapsed: {end_time - start_time}")
        print(f"Good Responses: {len(good_responses)}")
        print(f"Bad Responses: {len(bad_responses)}")

        if bad_responses:
            print("Bad Status Codes or Errors:")
            for error in set(bad_responses):
                print(error)

asyncio.run(main())

AIOHTTP 仅用了 4 秒多就极速完成任务。此 HTTP 客户端每秒生成超过 241 个请求！AIOHTTP 的性能大约是 Requests 的 10 倍，比 HTTPX 快了将近 50%。在 Python 中，AIOHTTP 的性能独领风骚。

Bright Data 的产品如何提供帮助

Bright Data 提供一系列解决方案，可增强您基于 HTTP 客户端的工作流程，尤其适用于数据密集型操作，例如网页抓取、API 请求和高性能集成。每款产品的适用情况如下：

住宅代理 – Bright Data 的住宅代理有助于在使用 Python HTTP 客户端（例如 AIOHTTP 或 HTTPX）抓取网站时避免屏蔽和被禁。这些代理模仿真实用户的行为，让您轻松访问受地理受限的内容或动态内容。
Web Scraper API – 您无需构建和维护自己的抓取基础架构，Bright Data 的 Web Scraper API 可让您通过预配置访问数百个热门网站。这使您可以专注于数据分析，无需操心请求处理、重试或被禁。只需使用 API 调用即可直接获取结构化数据。
现成的数据集 – 对于那些需要特定数据点但又想完全避免抓取操作的人而言，Bright Data 提供根据您的需求量身定制的现成数据集。这些数据集包括产品详情、价格和评价，可立即用于电商分析或市场调研。
Web Unlocker – Web Unlocker 可自动处理验证码、反机器人机制和复杂请求模式等挑战。将其与 HTTPX 或 AIOHTTP 等库搭配使用，可以简化难访问网站的抓取过程。
SERP API – 如果您要从搜索引擎提取数据，Bright Data 的 SERP API 可以简化提取过程，让您实时、可靠地访问搜索结果、广告和排名，无需担心基础架构或屏蔽。

将 Bright Data 的工具与 Python HTTP 客户端集成后，您可以构建强大、高性能的系统。这些系统能够简化数据收集，同时克服网络抓取和数据采集过程中的典型挑战。

结语

在 HTTP 客户端领域中，Requests 因易于使用而成为标准。与 Requests 相比，HTTPX 更像是从马车升级为现代汽车。它在高性能和易用性之间达到平衡。AIOHTTP 则像一艘火箭飞船。除非绝对必要，否则我们不会使用 AIOHTTP，但它仍然是迄今为止市面上最快的 HTTP 客户端。

立即注册 Bright Data，解锁强大的数据解决方案并获得您的企业所需

免费试用

Requests、HTTPX 与AIOHTTP：详细比较

简要概述

Python Requests

HTTPX

AIOHTTP

性能对比

Requests

HTTPX

AIOHTTP

Bright Data 的产品如何提供帮助

结语

你也可能对此有兴趣

如何使用 Python 抓取 Google 图片

如何抓取财经数据

如何结合使用代理和 HTTPX