尧图网站建设 
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对应关系很清楚。先看同步版有多慢python991234567891011import requestsurls [fhttps://example.com/page/{i} for i in range(20)]def fetch(url):resp requests.get(url)return resp.textfor url in urls:fetch(url)20个请求挨个来每个假设1秒总共20秒起步。再看异步版python991234567891011121314151617import asyncioimport aiohttpurls [fhttps://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来控制并发数。python991234567891011121314151617181920import asyncioimport aiohttpMAX_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 [fhttps://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_exceptionspython912results await asyncio.gather(*tasks, return_exceptionsTrue)这样异常会变成结果返回不会中断全部任务。然后你再过滤出成功的就行。方式二在单个任务里try-catch重试这个更实用给你一个带重试的模板python991234567891011121314async def fetch_with_retry(session, url, semaphore, max_retries3):for attempt in range(max_retries):try:async with semaphore:async with session.get(url, timeoutaiohttp.ClientTimeout(total10)) 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 Noneawait asyncio.sleep(2 ** attempt) # 指数退避return None核心逻辑失败了就等一会儿再试每次等待时间翻倍1秒、2秒、4秒这叫指数退避。网络抖动、临时限流都能扛过去。生产环境里没有重试机制的异步请求都是耍流氓。六、第四步异步数据库操作别让数据库成为新瓶颈很多人把网络请求改成异步了结果数据库还是同步操作。忙了半天瓶颈从网络转移到数据库了。SQLite可以用aiosqliteMySQL用aiomysqlPostgreSQL用asyncpg。我以SQLite为例python99123456789101112131415161718192021222324252627282930import asyncioimport aiosqliteasync 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计时python91234567import timestart time.perf_counter()# 你的异步代码end time.perf_counter()print(f耗时: {end - start:.2f}秒)同步版和异步版各跑一遍对比一下数字最有说服力。2. 逐步增加并发找到最佳值从10开始20、50、100逐步往上加看总耗时的变化。到了某个点之后耗时不再明显下降甚至开始上升那就是临界点了。3. 用asyncio的调试模式python9123import asyncioasyncio.run(main(), debugTrue)debug模式下协程跑得慢了会有警告能帮你找到阻塞点。4. CPU密集型任务别硬上异步这一点特别重要异步只对IO密集型任务网络请求、文件读写、数据库操作有效。如果你的代码主要在做计算异步不会带来任何提升反而因为调度开销更慢。CPU密集型怎么办用多进程concurrent.futures.ProcessPoolExecutor了解一下。八、新手最容易踩的5个坑在async函数里调用同步阻塞代码比如在async函数里用requests.get()、time.sleep()这会把整个事件循环堵住异步直接废了。记住网络用aiohttpsleep用asyncio.sleep。忘记await调用协程函数不加await它不会执行只会返回一个coroutine对象还会给你个RuntimeWarning。gather里一个失败全挂掉前面说了要么加return_exceptionsTrue要么每个任务自己处理异常。并发数设太高把对方服务器搞崩了或者自己被封IP。先用小并发试逐步增加。所有代码都想异步化没必要。简单脚本、一次性任务同步写就挺好。异步适合IO多、量大的场景为了异步而异步只会增加复杂度。这些坑你中过几个九、总结一下异步编程没那么神秘核心就是一句话别傻等等待的时间干点别的。今天讲了五步用aiohttp替换requests把同步请求改成异步用信号量控制并发避免被封IP或压垮服务器加上重试和异常处理让程序更健壮数据库也换成异步驱动打通整个链路压测调优找到最佳并发数按照这五步来你的脚本跑不快才怪。我自己写爬虫、做数据采集、跑批量任务都是这套流程屡试不爽。当然本地开发调试方便但真要让脚本7×24小时跑起来还得有个稳定的运行环境。这方面你可以了解下 VicroCode - AI智能体开发与Web应用托管平台 | HTML在线运行/Python在线运行支持Python应用在线部署和托管定时任务、API接口都能搞定不用自己买服务器、配环境开箱即用。你最近有什么项目想改异步的评论区聊聊说不定下次就写你的案例。我整理了一些市场信息和学习资料AI行业动态_AI编程实战案例_独立开发者资讯 - VicroCode 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对应关系很清楚。先看同步版有多慢python991234567891011import requestsurls [fhttps://example.com/page/{i} for i in range(20)]def fetch(url):resp requests.get(url)return resp.textfor url in urls:fetch(url)20个请求挨个来每个假设1秒总共20秒起步。再看异步版python991234567891011121314151617import asyncioimport aiohttpurls [fhttps://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来控制并发数。python991234567891011121314151617181920import asyncioimport aiohttpMAX_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 [fhttps://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_exceptionspython912results await asyncio.gather(*tasks, return_exceptionsTrue)这样异常会变成结果返回不会中断全部任务。然后你再过滤出成功的就行。方式二在单个任务里try-catch重试这个更实用给你一个带重试的模板python991234567891011121314async def fetch_with_retry(session, url, semaphore, max_retries3):for attempt in range(max_retries):try:async with semaphore:async with session.get(url, timeoutaiohttp.ClientTimeout(total10)) 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 Noneawait asyncio.sleep(2 ** attempt) # 指数退避return None核心逻辑失败了就等一会儿再试每次等待时间翻倍1秒、2秒、4秒这叫指数退避。网络抖动、临时限流都能扛过去。生产环境里没有重试机制的异步请求都是耍流氓。六、第四步异步数据库操作别让数据库成为新瓶颈很多人把网络请求改成异步了结果数据库还是同步操作。忙了半天瓶颈从网络转移到数据库了。SQLite可以用aiosqliteMySQL用aiomysqlPostgreSQL用asyncpg。我以SQLite为例python99123456789101112131415161718192021222324252627282930import asyncioimport aiosqliteasync 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计时python91234567import timestart time.perf_counter()# 你的异步代码end time.perf_counter()print(f耗时: {end - start:.2f}秒)同步版和异步版各跑一遍对比一下数字最有说服力。2. 逐步增加并发找到最佳值从10开始20、50、100逐步往上加看总耗时的变化。到了某个点之后耗时不再明显下降甚至开始上升那就是临界点了。3. 用asyncio的调试模式python9123import asyncioasyncio.run(main(), debugTrue)debug模式下协程跑得慢了会有警告能帮你找到阻塞点。4. CPU密集型任务别硬上异步这一点特别重要异步只对IO密集型任务网络请求、文件读写、数据库操作有效。如果你的代码主要在做计算异步不会带来任何提升反而因为调度开销更慢。CPU密集型怎么办用多进程concurrent.futures.ProcessPoolExecutor了解一下。八、新手最容易踩的5个坑在async函数里调用同步阻塞代码比如在async函数里用requests.get()、time.sleep()这会把整个事件循环堵住异步直接废了。记住网络用aiohttpsleep用asyncio.sleep。忘记await调用协程函数不加await它不会执行只会返回一个coroutine对象还会给你个RuntimeWarning。gather里一个失败全挂掉前面说了要么加return_exceptionsTrue要么每个任务自己处理异常。并发数设太高把对方服务器搞崩了或者自己被封IP。先用小并发试逐步增加。所有代码都想异步化没必要。简单脚本、一次性任务同步写就挺好。异步适合IO多、量大的场景为了异步而异步只会增加复杂度。这些坑你中过几个九、总结一下异步编程没那么神秘核心就是一句话别傻等等待的时间干点别的。今天讲了五步用aiohttp替换requests把同步请求改成异步用信号量控制并发避免被封IP或压垮服务器加上重试和异常处理让程序更健壮数据库也换成异步驱动打通整个链路压测调优找到最佳并发数按照这五步来你的脚本跑不快才怪。我自己写爬虫、做数据采集、跑批量任务都是这套流程屡试不爽。当然本地开发调试方便但真要让脚本7×24小时跑起来还得有个稳定的运行环境。这方面你可以了解下 VicroCode - AI智能体开发与Web应用托管平台 | HTML在线运行/Python在线运行支持Python应用在线部署和托管定时任务、API接口都能搞定不用自己买服务器、配环境开箱即用。你最近有什么项目想改异步的评论区聊聊说不定下次就写你的案例。我整理了一些市场信息和学习资料AI行业动态_AI编程实战案例_独立开发者资讯 - VicroCode