首页 > 脚本专栏 > python > Python异步请求

Python异步IO请求入门之网络请求提速技巧分享

2026-06-28 09:10:49 作者：detayun

本文详细讲解了Python异步IO从零入门,重点介绍aiohttp实现并发网络请求,覆盖协程基础概念、单条GET请求、并发请求多个URL、性能对比测试等实用内容,希望对大家有所帮助

前言

做接口调用、批量数据抓取时，大家最常用的是requests库。但requests是同步阻塞请求：发起请求后，程序会原地卡死，必须等待服务器返回结果才能执行下一行代码。如果要请求100个接口，串行等待会耗费大量时间。

而Python异步IO可以做到：等待网络响应时，程序不发呆，自动切换去执行其他请求。在大量IO等待的场景下，效率可以提升几倍甚至几十倍。

本文从零入门，只讲最实用的内容：先搞懂协程基础，再用asyncio + aiohttp实现异步网络请求，最后写出可直接运行的并发爬虫代码。

运行环境：Python 3.7及以上（支持async/await关键字）。

一、核心基础概念

1. 两个关键字

async def：用来定义协程函数。普通函数调用会直接执行，协程函数调用只会生成一个协程对象，代码不会立刻运行。
await：用来挂起当前协程。遇到网络等待时，让出CPU，事件循环自动去执行其他任务，等待IO完成后再恢复运行当前代码。

2. 事件循环（EventLoop）

异步程序的调度中枢，负责调度多个协程。在新版本Python中，统一使用asyncio.run()来启动事件循环，不用手动创建循环。

3. 异步请求库：aiohttp

requests只支持同步，异步HTTP请求首选aiohttp，它完全适配asyncio协程模型，内置连接池，是目前行业主流方案。

安装命令：

pip install aiohttp

二、第一个异步请求：单条GET请求

同步写法（requests）：

import requests

def sync_request():
    resp = requests.get("https://httpbin.org/get")
    print(resp.status_code)
    print(resp.text)

sync_request()

异步写法（aiohttp）：

import asyncio
import aiohttp

# 定义协程函数
async def async_request():
    # 创建会话（复用连接，不要频繁新建）
    async with aiohttp.ClientSession() as session:
        # 发起GET请求
        async with session.get("https://httpbin.org/get") as resp:
            print("状态码：", resp.status)
            # 读取响应必须加await，这是IO操作
            text = await resp.text()
            print(text[:200])

# 启动事件循环，运行协程
if __name__ == "__main__":
    asyncio.run(async_request())

新手必踩坑

协程函数不能直接调用执行，必须交给asyncio.run()；
读取响应text()、json()、read()都是IO操作，前面必须写await；
优先使用async with上下文管理器，自动关闭会话与连接，避免连接泄漏。

三、真正的并发：一次性请求多个URL

单条请求看不出优势，批量请求才是异步的主战场。

我们用asyncio.create_task()创建任务，再用asyncio.gather()收集所有任务结果，实现并发执行。

示例代码

import asyncio
import aiohttp

# 待请求的地址列表
url_list = [
    "https://httpbin.org/get",
    "https://httpbin.org/ip",
    "https://httpbin.org/headers"
]

async def fetch(session, url):
    """单个请求任务"""
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        return await resp.json()

async def main():
    tasks = []
    async with aiohttp.ClientSession() as session:
        for url in url_list:
            # 创建异步任务
            task = asyncio.create_task(fetch(session, url))
            tasks.append(task)
        # 等待所有任务完成，收集全部返回值
        results = await asyncio.gather(*tasks)
    for res in results:
        print(res)

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

运行效果：3个请求几乎同时发出，总耗时只等于最慢的那一次网络请求，而串行请求总耗时是三次请求时间之和。

四、常用进阶用法

1. POST请求（传递JSON数据）

async def post_json(session):
    data = {"username": "test", "password": "123456"}
    async with session.post(
        url="https://httpbin.org/post",
        json=data
    ) as resp:
        return await resp.json()

2. 限制并发数量（防止请求过猛被封IP）

无限并发会瞬间打爆目标服务器，还容易触发反爬。可以用信号量Semaphore限制同时运行的请求数量：

import asyncio
import aiohttp

# 最大并发数
sem = asyncio.Semaphore(5)

async def limited_fetch(session, url):
    async with sem:  # 信号量自动控制并发
        async with session.get(url) as resp:
            return await resp.status

3. 配置请求头、超时、连接池

# 连接池配置，限制最大连接数
connector = aiohttp.TCPConnector(limit=50)
# 全局超时设置
timeout = aiohttp.ClientTimeout(total=15)
# 请求头
headers = {"User-Agent": "Mozilla/5.0"}

async def main():
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers=headers
    ) as session:
        pass

五、异步 vs 同步：性能对比测试

串行同步代码

import time
import requests

url = "https://httpbin.org/get"
start = time.time()
for _ in range(10):
    requests.get(url)
print(f"同步串行耗时：{time.time()-start:.2f}s")

异步并发代码

import time
import asyncio
import aiohttp

url = "https://httpbin.org/get"

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

async def main():
    tasks = []
    async with aiohttp.ClientSession() as session:
        for _ in range(10):
            tasks.append(asyncio.create_task(fetch(session)))
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(main())
print(f"异步并发耗时：{time.time()-start:.2f}s")

测试结论：同步10次请求总耗时约5_{8秒；异步并发仅需0.5}1.5秒，性能差距一目了然。

补充：异步只优化IO等待，不会提升CPU密集型代码的运行速度。CPU计算任务依然要用多进程。

六、新手常见问题总结

RuntimeWarning: coroutine was never awaited：协程没有被调度执行，必须放到create_task或者await里，不能只调用函数。

事件循环嵌套报错：在Jupyter Notebook里会出现嵌套循环问题，本地脚本直接用asyncio.run()不会出错。

不要反复新建ClientSession：会话会维护TCP连接池，全局只创建一个Session，性能最高。

必须捕获异常：网络请求容易超时、连接失败，建议在fetch函数内部增加try-except捕获异常，避免整个程序崩溃。

七、完整健壮版模板（可直接用于爬虫）

import asyncio
import aiohttp

# 配置
MAX_CONCURRENT = 10
TIMEOUT = aiohttp.ClientTimeout(total=10)
URLS = [f"https://httpbin.org/get?id={i}" for i in range(20)]

sem = asyncio.Semaphore(MAX_CONCURRENT)

async def request_one(session, url):
    async with sem:
        try:
            async with session.get(url, timeout=TIMEOUT) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    return f"异常状态码：{resp.status}"
        except Exception as e:
            return f"请求失败：{url} | {str(e)}"

async def run_all():
    tasks = []
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT)
    async with aiohttp.ClientSession(connector=connector) as session:
        for url in URLS:
            task = asyncio.create_task(request_one(session, url))
            tasks.append(task)
        results = await asyncio.gather(*tasks)
    for item in results:
        print(item)

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

结尾

异步请求是Python爬虫、批量接口调用的必备技能。入门阶段牢牢抓住三点：

用async def写协程，IO操作加await；
使用aiohttp.ClientSession管理连接；
通过gather + task实现并发，用信号量控制请求频率。

掌握本篇内容后，你就可以轻松写出高并发网络程序，彻底摆脱同步阻塞带来的低效问题。后续可以继续学习异步MySQL、Redis，搭建完整的异步服务。

到此这篇关于Python异步IO请求入门之网络请求提速技巧分享的文章就介绍到这了,更多相关Python异步请求内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！