Python生态HTTP客户端类库：requests、httpx、aiohttp、Niquests、httpcore

johnny233

519人浏览 · 2026-03-21 10:53:36

johnny233 · 2026-03-21 10:53:36 发布

本文汇总一些Python生态下的HTTP客户端库。

库名称	特点	优点	缺点
`requests`	简单易用的HTTP库，基于`urllib3`	- 语法简洁 - 社区支持强大 - 易于上手和维护	- 阻塞式调用，不支持异步操作 - 相比`aihttp`体积较大
`http.client`	Python标准库中的低级HTTP库	- 无需安装第三方库 - 提供底层访问，可自定义程度高	- API使用相对复杂 - 缺少高级HTTP功能
`aiohttp`	异步的HTTP网络通信库，支持HTTP/1.1和HTTP/2	- 支持异步操作，适合高并发 -支持WebSockets	- 异步编程模型学习曲线陡峭 - 较新，社区支持不如`requests`
`urllib`	Python标准库，提供URL处理	- 无需安装第三方库 - 功能全面，包括请求和错误处理	- 易用性不如`requests` - 不支持异步操作
`httpx`	支持HTTP/1.1和HTTP/2的异步HTTP库	- 支持同步和异步请求 - 支持HTTP/2 - 可扩展性好	- 相对于`requests`，知名度和社区支持较小
`treq`	基于`Twisted`的异步HTTP客户端，使用`requests`的API风格	- 异步操作 - 与`requests`类似的API - 适用于`Twisted`用户	- 依赖于`Twisted`框架 - 社区支持有限
`requests-toolbelt`	`requests`的官方扩展，提供额外功能	- 增加`requests`没有的功能 - 流式上传下载支持	- 作为扩展，需要与`requests`结合使用 - 功能较为特定

requests

开源（GitHub，53.9K Star，9.8K Fork），Python生态下最老牌库。官方文档。

requests库不支持HTTP/2的主要原因：

架构设计：requests库基于urllib3构建，而urllib3本身是为HTTP/1.1设计
二进制帧层：HTTP/2使用二进制帧传输，而HTTP/1.1使用文本格式，两者在底层实现上有根本差异
多路复用：HTTP/2支持多路复用，这需要完全不同的连接管理机制

aiohttp

基于Python生态asyncio库构建的开源（GitHub，16.4K Star，2.2K Fork）异步HTTP客户端和服务器框架，能够同时处理数千个并发连接而不会阻塞主线程。与传统的同步请求库（如requests）相比，aiohttp在需要发起大量HTTP请求的场景下性能优势明显。该库同时支持客户端和服务器端功能，既可以用于爬虫、API调用等客户端场景，也可构建高性能的异步Web服务。官方文档。

实战

安装：pip install aiohttp

特性：

完全异步：基于asyncio，支持高并发非阻塞IO操作
客户端与服务器：同时提供HTTP客户端和Web服务器功能
WebSocket支持：原生支持WebSocket双向通信
中间件机制：灵活的中间件系统，便于扩展功能
会话管理：高效的连接池和会话复用机制
流式传输：支持大文件的流式上传和下载

使用aiohttp发起异步GET请求，通过async/await语法实现非阻塞调用。ClientSession对象管理连接池，能够复用TCP连接提高效率：

import aiohttp
import asyncio

async def fetch_data(url):
	async with aiohttp.ClientSession() as session:
		async with session.get(url) as response:
			print(f"状态码: {response.status}")
			data = await response.text()
			return data

# 运行异步任务
async def main():
	url = 'https://api.github.com'
	result = await fetch_data(url)
	print(f"响应内容长度: {len(result)}")

asyncio.run(main())

使用asyncio.gather同时发起多个HTTP请求，所有请求并行执行：

async def fetch_multiple(urls):
	async with aiohttp.ClientSession() as session:
		tasks = []
		for url in urls:
			task = asyncio.create_task(session.get(url))
			tasks.append(task)
		responses = await asyncio.gather(*tasks)
		results = []
		for response in responses:
			data = await response.text()
			results.append(data)
		return results

async def main():
	urls = [
		'https://api.github.com',
		'https://httpbin.org/get',
		'https://jsonplaceholder.typicode.com/posts/1'
	]
	results = await fetch_multiple(urls)
	print(f"成功获取 {len(results)} 个响应")

asyncio.run(main())

使用aiohttp发送POST请求，包括JSON数据的提交和响应的解析。通过json参数可自动序列化Python字典为JSON格式，response.json()则将响应反序列化为Python对象。

async def post_data():
	async with aiohttp.ClientSession() as session:
		url = 'https://httpbin.org/post'
		data = {
			'username': 'testuser',
			'email': 'test@example.com'
		}

		async with session.post(url, json=data) as response:
			result = await response.json()
			print(f"服务器响应: {result['json']}")

asyncio.run(post_data())

aiohttp不仅是客户端工具，还能构建高性能的异步Web服务器。创建一个简单的Web应用，定义路由和处理函数。通过装饰器或路由表配置URL映射，处理函数使用async def定义以支持异步操作：

from aiohttp import web

async def handle_index(request):
	return web.Response(text="欢迎访问首页")

async def handle_api(request):
	data = {'message': 'API响应', 'status': 'success'}
	return web.json_response(data)

# 创建应用和路由
app = web.Application()
app.router.add_get('/', handle_index)
app.router.add_get('/api', handle_api)

if __name__ == '__main__':
	web.run_app(app, host='127.0.0.1', port=8080)

通过aiohttp.ClientTimeout可精细控制连接、读取等各阶段的超时时间，捕获各种异常，可完美解决生产环境的网络异常和超时情况。

async def fetch_with_timeout():
	timeout = aiohttp.ClientTimeout(total=10, connect=3)
	async with aiohttp.ClientSession(timeout=timeout) as session:
		try:
			async with session.get('https://httpbin.org/delay/2') as response:
				data = await response.text()
				print("请求成功")
		except asyncio.TimeoutError:
			print("请求超时")
		except aiohttp.ClientError as e:
            print(f"请求错误: {e}")

asyncio.run(fetch_with_timeout())

WebSocket提供全双工通信能力，适合实时应用，aiohttp的WebSocket客户端用法，建立连接后可以持续发送和接收消息。

async def websocket_example():
	async with aiohttp.ClientSession() as session:
		async with session.ws_connect('wss://echo.websocket.org') as ws:
			await ws.send_str('Hello WebSocket')
			# 接收消息
			async for msg in ws:
				if msg.type == aiohttp.WSMsgType.TEXT:
					print(f"收到消息: {msg.data}")
					break
				elif msg.type == aiohttp.WSMsgType.ERROR:
					print("WebSocket错误")
					break

asyncio.run(websocket_example())

httpx

官网，Encode团队开发并开源（GitHub，15.2K Star，1.1K Fork）的新一代Python网络请求库，结合requests库的易用性和现代HTTP客户端的强大功能，旨在提供更高效、更灵活的HTTP请求处理能力。

核心特性：

异步支持：内置异步API，在高并发场景下能够显著提升性能。与传统requests库相比，可避免阻塞I/O操作，提高吞吐量；
HTTP/2支持：原生支持HTTP/2协议，提供更高效的网络请求处理，包括多路复用、头部压缩和服务器推送等功能；
连接池：减少创建新连接的开销，提升请求性能；
代理设置和认证机制：方便与需要认证的网络服务进行交互；
上下文管理器：可确保资源正确释放；
API兼容性：httpx的设计与requests库非常相似，从requests迁移到httpx变得相对容易。大多数requests的代码只需稍作修改就能在httpx上运行。

对比requests库，在内部实现和功能支持上有显著差异：

异步支持：httpx内置异步API，而requests是纯同步的
HTTP/2支持：httpx原生支持HTTP/2，requests不支持
类型注解：httpx有完整的类型注解，便于静态类型检查
超时处理：httpx提供更严格的超时控制

适用场景：

需要处理大量并发HTTP请求的应用
需要HTTP/2支持的项目
现代异步Python应用
需要从requests迁移但希望获得更好性能的项目

实战

安装：pip install httpx

如果需要HTTP/2支持，则需要额外安装：pip install httpx[http2]。

入门案例：

import httpx

response = httpx.get('https://httpbin.org/get')
print(response.status_code)
print(response.text) # 响应内容

data = {'key': 'value'}
response = httpx.post('https://httpbin.org/post', json=data)
print(response.json()) # 解析JSON响应

headers = {'User-Agent': 'my-app/1.0.0'}
response = httpx.get('https://httpbin.org/headers', headers=headers)

异步请求：

import httpx
import asyncio

async def main():
	async with httpx.AsyncClient() as client:
		response = await client.get('https://httpbin.org/get')

asyncio.run(main())

Niquests

Python生态下全新的开源（GitHub，1.7K Star，42 Fork）高性能HTTP客户端，目标是成为requests替代品，提供异步支持和性能优化；最大优势是API与requests高度兼容。官方文档。

实战

安装：pip install niquests

入门案例：

import niquests as requests # 直接替代import requests

response = requests.get(‘https://httpbin.org/get’)
print(f‘请求状态码: {response.status_code}’)
print(f‘响应内容类型: {response.headers[“content-type”]}’)
print(f‘响应体长度: {len(response.content)} 字节’)

原生支持异步操作，性能远超同步requests

import asyncio
import niquests

async def fetch_multiple_pages():
	urls = [
		‘https://httpbin.org/delay/1’，
		‘https://httpbin.org/delay/2’，
	]
    async with niquests.AsyncSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        for i, resp in enumerate(responses):
            print(f‘URL {i} 耗时: {resp.elapsed.total_seconds():.2f}秒’)

# 运行异步函数
asyncio.run(fetch_multiple_pages())

连接池、HTTP/2与更精细的控制。轻松创建可复用的Session对象来管理Cookie和连接，优化多次请求的性能。

session = niquests.Session(
	pool_connections=10，
	pool_maxsize=10，
	timeout=niquests.Timeout(connect=5.0, read=30.0)
)
# 启用HTTP/2支持（如果服务器支持）
session.http2 = True

response = session.get(‘https://httpbin.org/json’)
print(f‘使用的HTTP协议版本: {response.http_version}’)
print(f‘本次连接是否复用: {response.reused}’)

内置强大的重试机制，允许自定义重试策略（如对特定状态码重试、设置退避间隔），比手动实现重试逻辑要可靠和简洁得多。

from niquests import Session
from urllib3.util import Retry

# 定义重试策略：总共重试3次，针对状态码500和超时错误
retry_strategy = Retry(
	total=3,
	status_forcelist=[500, 502, 503, 504]，
	allowed_methods=[“GET”, “POST”]
)
# 将重试策略适配到会话
session = Session()
session.mount(‘https://’， niquests.adapters.HTTPAdapter(max_retries=retry_strategy))

try:
	resp = session.get(‘https://httpbin.org/status/500’)
	print(f‘最终状态码: {resp.status_code}’)
except Exception as e:
	print(f‘请求最终失败: {type(e).__name__}’)

对比

与requests相比，提供原生异步支持和显著的性能提升，且API兼容；
与httpx相比，在API设计上对requests用户更友好，迁移成本几乎为零；
与aiohttp相比，提供更同步风格的异步API，学习曲线更平缓。

不足在于诞生时间较短，社区生态和第三方适配尚在发展。

建议：任何正在使用requests并希望获得异步能力或性能提升的项目，都可考虑迁移到Niquests。

httpcore

和httpx一样，也是Encode团队开发并开源（GitHub，533 Star，152 Fork），专注于底层HTTP通信的Python库，它是httpx库的核心传输层实现。与高层HTTP客户端库不同，httpcore提供了最小化的HTTP/1.1和HTTP/2协议实现。设计理念是提供一个稳定、高效的HTTP传输层，让其他库可以在此基础上构建更高级的功能。官方文档。

功能特性：

HTTP/1.1和HTTP/2支持：完整实现两种协议，自动协商使用最优协议
连接池管理：高效的连接复用机制，减少连接建立开销
异步和同步API：提供async和sync两种接口，适应不同场景；请求发送&响应接收
流式传输：支持请求和响应的流式处理，适合大文件传输
超时控制：精细的超时配置，包括连接超时、读取超时等
自动重试：
代理支持：支持HTTP和SOCKS代理配置
TLS/SSL支持：完整的HTTPS支持，可自定义SSL配置

实战

安装：pip install httpcore

httpcore提供底层的HTTP请求接口，通过ConnectionPool管理连接。与高层库不同，httpcore需要手动管理连接池的生命周期，在发送请求时，需要指定完整的URL组件，包括scheme、host、port等。这种显式的API设计使得代码逻辑更加清晰，便于调试和优化。

import httpcore

pool = httpcore.ConnectionPool()
response = pool.request(
	method=b"GET",
	url=(b"https", b"www.example.com", 443, b"/"),
)
status = response.status
headers = response.headers
content = response.content
print(f"响应内容: {content[:100]}")
pool.close()

httpcore完整支持异步编程，提供AsyncConnectionPool用于异步场景。异步API在处理高并发请求时具有显著优势，可在等待IO操作时执行其他任务，大幅提升程序效率。异步连接池的使用方式与同步版本类似，但需要使用async/await语法，异步模式特别适合需要同时处理大量HTTP请求的场景，如爬虫、API网关等应用。

import httpcore
import asyncio

async def fetch_data():
    # 创建异步连接池
    async with httpcore.AsyncConnectionPool() as pool:
        # 发送异步请求
        response = await pool.request(
            method=b"GET",
            url=(b"https", b"httpbin.org", 443, b"/get"),
        )
        status = response.status
        content = response.content
        print(f"响应: {content[:100]}")

asyncio.run(fetch_data())

httpcore支持流式读取响应内容，在处理大文件下载或实时数据流时非常有用，通过stream方法可逐块读取响应数据，避免一次性加载整个响应到内存，节省内存，还能更快地开始处理数据。流式处理特别适合下载大文件、处理视频流、实时日志收集等场景。

import httpcore

pool = httpcore.ConnectionPool()
response = pool.request(
    method=b"GET",
    url=(b"https", b"httpbin.org", 443, b"/stream/10"),
)
print("流式读取响应:")
for chunk in response.iter_stream():
    print(f"接收数据块: {len(chunk)} 字节")
    # 处理数据块
    if chunk:
        print(chunk[:50])

pool.close()

httpcore原生支持HTTP/2协议，可利用多路复用特性在单个连接上并发处理多个请求。HTTP/2相比HTTP/1.1具有更高的传输效率，减少延迟和带宽消耗。httpcore会自动检测服务器是否支持HTTP/2，并在可能的情况下使用该协议。多路复用特性使得可在一个TCP连接上同时发送多个请求，避免队头阻塞问题。

import httpcore

pool = httpcore.ConnectionPool(http2=True)
# 发送多个请求，利用HTTP/2多路复用
urls = [
	(b"https", b"www.example.com", 443, b"/"),
	(b"https", b"www.example.com", 443, b"/about"),
	(b"https", b"www.example.com", 443, b"/contact"),
]

for url in urls:
    response = pool.request(method=b"GET", url=url)
    print(f"请求 {url[3]}: 状态码 {response.status}")

pool.close()

httpcore提供丰富的连接池配置选项，可以根据应用需求调整连接行为。通过设置最大连接数、连接超时、保持连接时间等参数，可以优化资源使用和性能表现。合理的连接池配置能够在高并发场景下保持稳定性，避免资源耗尽。

import httpcore

# 创建自定义配置连接池
pool = httpcore.ConnectionPool(
    max_connections=100,  # 最大连接数
    max_keepalive_connections=20,  # 保持活跃的连接数
    keepalive_expiry=5.0,  # 连接保持时间（秒）
    http2=True,  # 启用HTTP/2
)
# 使用配置好的连接池
response = pool.request(
    method=b"GET",
    url=(b"https", b"httpbin.org", 443, b"/get"),
)
pool.close()