GRequests - 异步请求

wangmcn

发布于 2025-11-18 09:32:39

810

1、前言

在 Python 开发中，处理大量 HTTP 请求时，传统的Requests库因串行执行效率低下，难以满足高并发场景需求。而GRequests库通过将Requests与Gevent的协程能力结合，实现了 HTTP 请求的异步并发，大幅提升了请求处理速度。

2、简介

GRequests 是一个结合了 Requests 和 Gevent 的 Python 库，旨在简化异步 HTTP请求的实现。

核心功能：

异步 HTTP 请求：基于 Gevent 实现并发，让 Requests 库支持异步操作，可同时发送多个 HTTP 请求，提高网络请求效率。
兼容 Requests API：提供与 Requests 相似的接口（如 get、post、put 等），支持相同的参数（如超时、请求头、钩子等），降低使用成本。
批量请求处理：通过 map()、imap() 等方法批量处理请求，支持设置并发数（size 参数）控制请求速率。
异常处理：允许自定义异常处理器，用于捕获超时、连接错误等异常，并可返回自定义结果。

使用pip安装：

pip install grequests

Github地址：

https://github.com/spyoungtech/grequests

3、快速上手

1、基本用法：

创建多个请求并批量发送，获取响应列表。

import grequests

urls = ['http://httpbin.org', 'http://www.heroku.com']
rs = (grequests.get(u) for u in urls)
responses = grequests.map(rs) # 并发发送请求

2、异常处理：

自定义处理器捕获请求失败。

def exception_handler(request, exception):
    print(f"Request to {request.url} failed: {exception}")

reqs = [grequests.get('http://fakedomain/', timeout=0.1)]
grequests.map(reqs, exception_handler=exception_handler)

3、生成器式处理：

使用 imap() 逐一生成响应（无序）。

for resp in grequests.imap((grequests.get(u) for u in urls), size=5):
    print(resp.status_code)

4、应用场景：

（1）批量数据采集与爬虫

场景：需要从多个网页或 API 接口抓取数据（如电商商品信息、新闻内容、股票数据等）。
优势：异步并发发送请求，大幅减少等待时间，提升爬取效率。例如，爬取 100 个页面时，无需逐个等待响应，可同时发起请求。
示例：批量获取多个网站的状态码或标题。

import grequests

urls = [f"https://example.com/page{i}" for i in range(1, 50)]
reqs = (grequests.get(url) for url in urls)
responses = grequests.map(reqs, size=10) # 控制并发数为10
for resp in responses:
    if resp:
        print(f"{resp.url}: {resp.status_code}")

（2）API批量调用与集成

场景：需要调用第三方 API 批量获取或提交数据（如批量查询地理位置、翻译文本、推送通知等）。
优势：减少 API 调用的总耗时，尤其适合对响应速度敏感的业务。例如，同时向多个服务节点发送请求，汇总结果后处理。
示例：批量查询多个IP的地理位置信息。

import grequests

ips = ["8.8.8.8", "1.1.1.1", "208.67.222.222"]
reqs = [grequests.get(f"http://ip-api.com/json/{ip}") for ip in ips]
results = grequests.map(reqs)
for res in results:
    if res:
        data = res.json()
        print(f"{data['query']}: {data['country']}")

（3）监控与健康检查

场景：对多个服务器、服务端点或接口进行定期健康检查（如检测是否存活、响应时间是否正常）。
优势：同时监控大量目标，快速获取整体状态，避免串行检查导致的延迟累积。
示例：批量检查服务可用性。

import grequests

services = [
    "https://api.serviceA.com/health",
    "https://api.serviceB.com/ping",
    "https://api.serviceC.com/status"
]

def handle_exception(req, exc):
    returnf"Failed to check {req.url}: {str(exc)}"

reqs = (grequests.get(url, timeout=2) for url in services)
statuses = grequests.map(reqs, exception_handler=handle_exception)
for status in statuses:
    print(status if isinstance(status, str) elsef"{status.url} is OK")

（4）并发文件下载

场景：需要从多个 URL 同时下载文件（如图片、文档、数据集等）。
优势：充分利用网络带宽，加速多文件下载过程，尤其适合大规模资源获取。
示例：批量下载图片。

import grequests
import os

image_urls = [
    "https://example.com/img1.jpg",
    "https://example.com/img2.png",
    "https://example.com/img3.gif"
]

def save_image(resp):
    if resp:
        filename = os.path.basename(resp.url)
        with open(filename, "wb") as f:
            f.write(resp.content)
        print(f"Saved {filename}")

reqs = [grequests.get(url, callback=save_image) for url in image_urls]
grequests.map(reqs)

（5）负载测试与压力测试

场景：模拟多个用户同时访问目标服务，测试其并发处理能力。
优势：通过控制并发数（`size` 参数），模拟不同压力场景，评估服务性能瓶颈。
示例：模拟 50 个并发请求测试接口响应。

import grequests
import time

target_url = "https://api.target.com/endpoint"
concurrency = 50  # 并发数
total_requests = 1000  # 总请求数

start = time.time()
reqs = (grequests.post(target_url, json={"data": i}) for i in range(total_requests))
grequests.map(reqs, size=concurrency)
print(f"Completed {total_requests} requests in {time.time() - start:.2f}s")

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-11-10，如有侵权请联系 cloudcommunity@tencent.com 删除

接口