Python在线运行异步编程高性能实战：5步让你的接口提速10倍-开发者社区

Python在线运行异步编程高性能实战：5步让你的接口提速10倍

上周帮朋友优化一个数据采集脚本，原来同步写的，跑一轮要3个多小时。我花了一个下午改成异步，最后20分钟跑完。他说我变魔术呢。

其实哪里是魔术，就是把同步换成异步而已。今天就把这套方法完整讲一遍，从原理到实战，看完你也能用。

一、先搞懂：异步到底快在哪？

很多人一听"异步编程"就觉得高大上，好像是什么黑科技。真不是。

你去咖啡店买咖啡，同步模式就是：你点单，站在柜台前等着，咖啡做好了拿走，然后下一个人才能点。异步模式就是：你点单，拿个号找位子坐，服务员做好了喊你，期间别人可以继续点单。

区别在哪？等待的时间能不能利用起来。

Python里大部分程序慢，不是CPU不够用，而是在等——等网络响应、等数据库返回、等文件读写。这些时间CPU其实闲着呢。异步就是把这些空闲时间利用起来，同时处理多个任务。

举个具体数字：你要请求100个网页，每个网页响应平均1秒。同步写的话，大概100秒。异步写呢？如果并发50个，理论上2秒多就搞定了。这就是10倍、50倍的差距来源。

懂了吧？不是代码跑得快了，是等待的时间被充分利用了。

你觉得你的项目里，哪些地方在傻等？

二、三个核心概念，搞明白就入门了

Python异步编程有三个东西必须搞清楚，不然代码写出来自己都看不懂。

1. 协程（coroutine）

就是用async def定义的函数。它跟普通函数的区别是：可以被暂停，也可以被恢复。遇到await的时候就暂停，去干别的事，等结果回来了再继续。

很像你烧水的时候，水没开你就去切菜，水开了再回来处理。

2. 事件循环（event loop）

它是整个异步程序的"调度中心"。哪个协程该运行了、哪个在等待、哪个等的结果回来了，全由它安排。Python 3.7之后用asyncio.run()就能自动创建和管理。

3. 任务（Task）

把协程包成Task，它才会被事件循环调度执行。如果只是定义了协程但没创建Task，它是不会自动跑的。这是新手最容易踩的坑。

一句话总结：协程是任务内容，事件循环是调度器，Task是把协程提交给调度器的方式。

这三个概念能分清吗？

三、第一步：用aiohttp替换requests

大多数人第一次接触异步，都是从网络请求开始的。同步用requests，异步用aiohttp，对应关系很清楚。

先看同步版有多慢：

python

import requests

urls = [f"https://example.com/page/{i}" for i in range(20)]

def fetch(url):

resp = requests.get(url)

return resp.text

for url in urls:

fetch(url)

20个请求挨个来，每个假设1秒，总共20秒起步。

再看异步版：

python

import asyncio

import aiohttp

urls = [f"https://example.com/page/{i}" for i in range(20)]

async def fetch(session, url):

async with session.get(url) as resp:

return await resp.text()

async def main():

async with aiohttp.ClientSession() as session:

tasks = [fetch(session, url) for url in urls]

results = await asyncio.gather(*tasks)

print(f"拿到了 {len(results)} 个结果")

asyncio.run(main())

两段代码结构其实差不多，关键就三步：函数前面加async，耗时操作前加await，最后用asyncio.gather并发执行。

如果你想动手试试，但又不想本地搭环境，可以直接在 python在线运行环境里跑，浏览器里就能写，不用装任何东西。

是不是看着也没那么难？

四、第二步：用信号量控制并发，别把服务器搞崩了

上一步的代码，20个请求还好。如果是200个、2000个呢？全发出去，对方服务器可能直接把你拉黑了。

这时候就需要信号量（Semaphore）来控制并发数。

python

import asyncio

import aiohttp

MAX_CONCURRENT = 10 # 最多同时10个请求

async def fetch(session, url, semaphore):

async with semaphore: # 拿到信号量才能执行

async with session.get(url) as resp:

return await resp.text()

async def main():

semaphore = asyncio.Semaphore(MAX_CONCURRENT)

async with aiohttp.ClientSession() as session:

urls = [f"https://example.com/page/{i}" for i in range(100)]

tasks = [fetch(session, url, semaphore) for url in urls]

results = await asyncio.gather(*tasks)

print(f"完成 {len(results)} 个请求")

asyncio.run(main())

原理很简单：信号量像一个有10把钥匙的房间，进去一个拿一把，出来一把放回去。钥匙用完了，后面的就得在门口等着。

并发数设多少合适？这个得看目标网站的承受能力。一般来说：

公共API：看文档里的限流要求
普通网站：10-20个比较安全
自己的服务器：可以大胆一点，50-100个

注意！注意！注意！并发不是越高越快。到一定程度后，网络带宽、对方服务器处理能力都会成为瓶颈，再往上加并发反而可能因为超时、重试变慢。

你之前踩过并发太高被封IP的坑吗？

五、第三步：异常处理与重试，别让一个失败拖垮全部

真实的网络环境里，超时、报错是常态。如果100个请求里有1个失败，asyncio.gather默认会直接全部报错。这肯定不行。

两种处理方式：

方式一：gather里加return_exceptions

python

results = await asyncio.gather(*tasks, return_exceptions=True)

这样异常会变成结果返回，不会中断全部任务。然后你再过滤出成功的就行。

方式二：在单个任务里try-catch+重试

这个更实用，给你一个带重试的模板：

python

async def fetch_with_retry(session, url, semaphore, max_retries=3):

for attempt in range(max_retries):

try:

async with semaphore:

async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:

resp.raise_for_status()

return await resp.text()

except Exception as e:

if attempt == max_retries - 1:

print(f"{url} 重试{max_retries}次都失败了: {e}")

return None

await asyncio.sleep(2 ** attempt) # 指数退避

return None

核心逻辑：失败了就等一会儿再试，每次等待时间翻倍（1秒、2秒、4秒），这叫指数退避。网络抖动、临时限流都能扛过去。

生产环境里，没有重试机制的异步请求都是耍流氓。

六、第四步：异步数据库操作，别让数据库成为新瓶颈

很多人把网络请求改成异步了，结果数据库还是同步操作。忙了半天，瓶颈从网络转移到数据库了。

SQLite可以用aiosqlite，MySQL用aiomysql，PostgreSQL用asyncpg。我以SQLite为例：

python

import asyncio

import aiosqlite

async def init_db():

async with aiosqlite.connect('data.db') as db:

await db.execute('''

CREATE TABLE IF NOT EXISTS articles (

id INTEGER PRIMARY KEY AUTOINCREMENT,

title TEXT,

content TEXT,

created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP

)

''')

await db.commit()

async def insert_article(title, content):

async with aiosqlite.connect('data.db') as db:

await db.execute(

'INSERT INTO articles (title, content) VALUES (?, ?)',

(title, content)

)

await db.commit()

async def main():

await init_db()

await insert_article("测试标题", "测试内容")

print("插入成功")

asyncio.run(main())

用法跟sqlite3几乎一样，就是每个操作前面加个await。

如果你用的是 VicroCode 平台，它自带SQLite数据库在线管理器，建表、查数据、改数据直接可视化操作，不用每次都写代码去查，省不少事。

数据库异步化之后，整个链路就通了：异步请求 → 异步处理 → 异步存储。全程不堵。

七、第五步：性能压测与调优

改完异步，怎么知道快了多少？哪些地方还能优化？

1. 用time.perf_counter计时

python

import time

start = time.perf_counter()

# 你的异步代码

end = time.perf_counter()

print(f"耗时: {end - start:.2f}秒")

同步版和异步版各跑一遍，对比一下，数字最有说服力。

2. 逐步增加并发，找到最佳值

从10开始，20、50、100逐步往上加，看总耗时的变化。到了某个点之后，耗时不再明显下降，甚至开始上升，那就是临界点了。

3. 用asyncio的调试模式

python

import asyncio

asyncio.run(main(), debug=True)

debug模式下，协程跑得慢了会有警告，能帮你找到阻塞点。

4. CPU密集型任务别硬上异步

这一点特别重要！异步只对IO密集型任务（网络请求、文件读写、数据库操作）有效。如果你的代码主要在做计算，异步不会带来任何提升，反而因为调度开销更慢。

CPU密集型怎么办？用多进程，concurrent.futures.ProcessPoolExecutor了解一下。

八、新手最容易踩的5个坑

在async函数里调用同步阻塞代码
比如在async函数里用requests.get()、time.sleep()，这会把整个事件循环堵住，异步直接废了。记住：网络用aiohttp，sleep用asyncio.sleep。
忘记await
调用协程函数不加await，它不会执行，只会返回一个coroutine对象，还会给你个RuntimeWarning。
gather里一个失败全挂掉
前面说了，要么加return_exceptions=True，要么每个任务自己处理异常。
并发数设太高
把对方服务器搞崩了，或者自己被封IP。先用小并发试，逐步增加。
所有代码都想异步化
没必要。简单脚本、一次性任务，同步写就挺好。异步适合IO多、量大的场景，为了异步而异步只会增加复杂度。

这些坑你中过几个？

九、总结一下

异步编程没那么神秘，核心就是一句话：别傻等，等待的时间干点别的。

今天讲了五步：

用aiohttp替换requests，把同步请求改成异步
用信号量控制并发，避免被封IP或压垮服务器
加上重试和异常处理，让程序更健壮
数据库也换成异步驱动，打通整个链路
压测调优，找到最佳并发数

按照这五步来，你的脚本跑不快才怪。我自己写爬虫、做数据采集、跑批量任务，都是这套流程，屡试不爽。

当然，本地开发调试方便，但真要让脚本7×24小时跑起来，还得有个稳定的运行环境。这方面你可以了解下 VicroCode - AI智能体开发与Web应用托管平台 | HTML在线运行/Python在线运行，支持Python应用在线部署和托管，定时任务、API接口都能搞定，不用自己买服务器、配环境，开箱即用。

你最近有什么项目想改异步的？评论区聊聊，说不定下次就写你的案例。

我整理了一些市场信息和学习资料：AI行业动态_AI编程实战案例_独立开发者资讯 - VicroCode