提到异步HTTP框架很多人的第一反应是“这东西我暂时用不上”——这很正常因为我以前也是这么想的。翻了翻这两年写的爬虫和接口服务真正卡住我的几乎都是同一个问题某个环节需要等网络IO程序却被同步逻辑拖得死死的要么排队等响应要么开一堆线程把内存吃满。后来认真把aiohttp用起来很多之前别扭的实现一下子顺了。aiohttp是一个基于asyncio的异步HTTP客户端/服务器框架简单说它让你在同一个线程里同时处理成百上千个网络请求不靠多线程也不靠多进程靠的是事件循环和协程的切换。这篇文章不会讲太多高深理论我尽量用实际能跑的代码、真实踩过的坑把一个完整的aiohttp从客户端到服务器端的用法串起来。不管你是写爬虫、写接口还是打算给自己的脚本加一个Web控制台这篇都应该对你有用。1. 从requests到aiohttp到底解决了什么痛点先说说我为什么放着好好的requests不用要折腾一个“异步版”。requests确实是Python社区里最好用的同步HTTP库之一API设计简洁文档友好我到现在仍然会在很多同步脚本里用它。但它有一个天生的限制当你调用requests.get()的时候程序会阻塞在那一行直到服务器返回响应或者超时。这个阻塞在单个请求的场景下感知不强顶多就是慢几百毫秒。可一旦你要抓几百个页面、调用几十个接口问题就来了——大部分时间其实都花在等网络IO上CPU基本是空闲的。用多线程可以缓解这个问题。之前我写过用concurrent.futures.ThreadPoolExecutor加requests的爬虫20个线程并发抓取速度确实提上来了但随之而来的问题也不少线程切换有开销、线程间共享数据结构要加锁、连接数一多还容易被服务器封IP。线程池不是不行只是总觉得“为了做个IO密集型任务把并发模型搞这么重不太优雅”。aiohttp的思路完全不同。它建立在asyncio之上请求发出后事件循环会挂起当前协程转而处理其他已经就绪的协程等响应回来了再切回来继续执行。整个过程单线程搞定没有线程切换的代价也不用手动管理线程池。用我自己的话总结多线程是“多个人各自排队等结果”aiohttp是“一个人同时盯着一堆请求谁先回来先处理谁”。另外一个容易被忽略的点是aiohttp不仅能当客户端还能当服务器。也就是说学一个库既能解决“主动发请求”的问题又能解决“对外提供接口”的问题。对于个人开发者或者小团队来说这确实是很划算的事情——你可以用同一个异步事件循环既向外抓数据又向内提供API服务。当然aiohttp也不是银弹。如果你是写CPU密集型任务比如图像处理、大量数值计算那该用多进程还是得用多进程异步帮不了你。aiohttp真正擅长的是IO密集型场景也就是“等网络、等磁盘、等数据库响应”这类任务。还要提醒一点异步编程的心智模型和同步编程真的不一样。不是装了aiohttp就能自动变快你得理解事件循环的调度方式才能在正确的地方让出控制权。下一节我会专门聊这个。2. 环境准备版本、虚拟环境和安装时那些容易翻车的细节aiohttp的安装本身很简单pip install aiohttp一行命令搞定。但基于我这些年踩过的坑有几个细节值得提前说清楚。2.1 Python版本怎么选aiohttp对Python版本有要求。目前最新版本要求Python 3.8以上如果你的Python是3.7或更老装上的aiohttp版本会旧一些某些新特性不可用。我的建议是直接用Python 3.10或3.11原因是asyncio在3.10之后有了不少体验上的改善比如asyncio.run()这个写法在3.7之后也稳定了整体写起来更顺手。怎么查当前Python版本终端里执行python --version如果你是第一次接触Python还没配好环境我建议先用pyenv或者官方安装包装一个干净的Python 3.10。极不建议直接装在系统自带的Python上尤其是macOS和Linux系统对Python的版本和site-packages有严格管理你直接pip install很容易遇到externally-managed-environment报错或者装上了但import的时候找不到模块。2.2 虚拟环境是必须的我在热词里看到很多人在问“python第三方库装了但import不到”十有八九是没有激活虚拟环境。很多人都有过这样的经历明明pip install xxx提示成功了进到项目里import xxx却报ModuleNotFoundError。原因很简单——你安装的Python解释器和运行代码的解释器不是同一个。所以每开一个新项目第一步永远是建虚拟环境python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate激活之后你的pip install就只会装到当前项目的虚拟环境里不会污染全局也不会跟系统Python打架。这套流程虽然啰嗦但能省去后面一大半的环境问题。2.3 国内源的加速配置从热词里能看到“python国内源”是一个高频搜索词。确实直接从PyPI官方源下载有时候慢到怀疑人生。我这边实测用清华源或者阿里源速度能提升好几倍。临时用的话在pip命令后面加-i参数pip install aiohttp -i https://pypi.tuna.tsinghua.edu.cn/simple如果希望一劳永逸可以改全局配置。在用户目录下创建一个~/.pip/pip.confLinux/macOS路径Windows是%APPDATA%\pip\pip.ini写入[global] index-url https://pypi.tuna.tsinghua.edu.cn/simple [install] trusted-host pypi.tuna.tsinghua.edu.cn2.4 安装后验证装完之后别急着开写先在终端里确认一下能不能正常导入python -c import aiohttp; print(aiohttp.__version__)如果能打印出版本号说明环境基本没问题。如果报错八成是虚拟环境没激活或者装到了别的解释器上回到2.2重新检查。3. 理解异步三件套协程、await和事件循环前面说了aiohttp是基于asyncio的如果不理解asyncio的几个核心概念写出来的代码很容易“看起来是异步跑起来是同步”。3.1 协程不是函数先做一个区分普通函数用def定义调用即执行协程用async def定义调用时不执行函数体而是返回一个协程对象必须等它被事件循环调度才会真正运行。async def say_hello(): return hello # 这里不会打印任何东西 coro say_hello() print(coro) # coroutine object say_hello at 0x...协程对象必须放进事件循环里才能跑。最简单的跑法import asyncio async def say_hello(): return hello asyncio.run(say_hello())asyncio.run()是Python 3.7提供的高层入口它负责创建事件循环、运行传入的协程、结束后关闭事件循环。绝大多数脚本入口都可以用这一行。3.2 await让出控制权的时机await是异步编程里最核心的关键字。它后面跟着的必须是一个可等待对象awaitable比如协程、asyncio.Task、asyncio.Future或者是实现了__await__方法的对象。真正重要的不是“await挂起协程”而是“await挂起的同时事件循环可以调度其他协程”。举个例子import asyncio async def request_a(): print(A start) await asyncio.sleep(2) # 模拟网络等待 print(A done) async def request_b(): print(B start) await asyncio.sleep(1) print(B done) async def main(): await asyncio.gather(request_a(), request_b()) asyncio.run(main())运行结果A start B start B done A done注意request_a()执行到await asyncio.sleep(2)时它把控制权交还给了事件循环于是事件循环立刻去跑request_b()。如果换成同步写法的sleep(2)加sleep(1)总耗时是3秒而这里总耗时只有2秒。这就是异步的核心价值。aiohttp的session.get()本质上和asyncio.sleep()一样都是IO等待操作所以它可以被await可以让出控制权。3.3 并发不是并行很多人刚接触异步时会误以为“异步多核并行”其实不是。asyncio是单线程的事件循环在一个线程里不断遍历就绪任务并执行它们。所谓并发指的是多个任务交替执行而并行才是指多个任务在不同核心上同时执行。aiohttp的并发属于前者。这个区别有什么实际意义意义在于如果你的任务里有大量CPU密集计算比如解析超大JSON、图像缩放这些计算会阻塞事件循环导致其他协程得不到调度。解决办法是把这类计算丢给线程池或进程池去执行或者缩小单次任务的计算量。总之别在async def函数里写死循环做大量计算。4. 客户端实操Session、并发请求和超时控制理论讲完来点真东西。先看一个最简单的GET请求怎么写。4.1 第一个aiohttp客户端请求import aiohttp import asyncio async def fetch_html(session, url): async with session.get(url) as resp: return await resp.text() async def main(): async with aiohttp.ClientSession() as session: html await fetch_html(session, https://www.example.com) print(html[:200]) asyncio.run(main())几个细节需要说明aiohttp.ClientSession()要放在async with里它会管理内部的连接池用完之后自动释放。session.get()返回的是一个ClientResponse对象它本身也是一个异步上下文管理器所以用async with包住确保响应正确关闭。resp.text()也是异步的要await。为什么读取响应体也是异步因为响应体是流式到达的边接收边解析如果同步读取就会阻塞事件循环。4.2 Session为什么不能为每个请求新建初学者最容易犯的错是每个请求都创建一个新的ClientSession。这样做会反复建立TCP连接和TLS握手性能上损失很大。正确做法是全局复用一个ClientSession它底层维护了一个连接池多个请求可以复用同一个TCP连接HTTP Keep-Alive。我实际写爬虫时一般都是把session作为参数传给各个协程或者放在自定义类属性里class Crawler: def __init__(self): self.session aiohttp.ClientSession() async def close(self): await self.session.close()4.3 并发抓取的正确姿势有了ClientSession和async with我们再加一点并发魔法。假设要抓10个URL用asyncio.gather把它们全部丢进去import aiohttp import asyncio import time URLS [fhttps://www.example.com/page/{i} for i in range(1, 11)] async def fetch_page(session, url): async with session.get(url) as resp: return await resp.text() async def main(): async with aiohttp.ClientSession() as session: tasks [fetch_page(session, url) for url in URLS] results await asyncio.gather(*tasks, return_exceptionsTrue) for url, result in zip(URLS, results): if isinstance(result, Exception): print(f{url} 失败: {result}) else: print(f{url} 成功, 长度{len(result)}) asyncio.run(main())注意两点tasks [fetch_page(session, url) for url in URLS]这行只是创建了协程对象并没有真正执行。真正开始执行是在await asyncio.gather(*tasks)这一行。return_exceptionsTrue这个参数很关键。如果某个请求抛了异常默认情况下gather会把整个异常抛出来导致后面的任务都拿不到结果。设成True异常会被当作返回值返回你可以逐个判断处理。4.4 超时控制不设置等于听天由命写爬虫或者调接口最怕的就是某个请求挂起不返回。不设置超时时间的话程序可能一直卡在那里。aiohttp的超时设置有两种方式。一种是全局设置timeout aiohttp.ClientTimeout(total30) async with aiohttp.ClientSession(timeouttimeout) as session: ...另一种是单个请求覆盖用asyncio.timeoutPython 3.11或asyncio.wait_for更通用async with session.get(url) as resp: html await asyncio.wait_for(resp.text(), timeout10)我个人习惯在创建ClientSession时就设置一个合理的total超时比如15秒避免某个慢接口拖垮整个并发任务。4.5 处理JSON响应和POST请求aiohttp处理JSON也很方便async def post_json(session, url, data): async with session.post(url, jsondata) as resp: return await resp.json()注意json参数会自动帮你做序列化并设置Content-Type: application/json。如果你传的是表单数据要用data参数async with session.post(url, data{key: value}) as resp: ...这里有个常见误区resp.json()是异步方法要await很多人从requests转过来时会忘掉这一点。4.6 请求头与Cookie携带实际应用中很多接口需要带请求头或者Cookie。aiohttp的ClientSession支持在创建时设置默认请求头也可以每个请求单独传headers { User-Agent: Mozilla/5.0 ..., Accept: application/json, } async with aiohttp.ClientSession(headersheaders) as session: # 所有请求默认带上上面的headers async with session.get(url) as resp: ...Cookie的话直接在请求时传cookies参数cookies {session_id: abc123} async with session.get(url, cookiescookies) as resp: ...5. 服务器端实操一个能跑的异步API服务aiohttp客户端说完再来看看它的服务器端能力。这部分的逻辑和Flask/FastAPI有点相似但由于和客户端共用同一个底层网络模型你可以在一个进程里既做爬虫又做接口服务数据共享起来很方便。5.1 最小可运行的aiohttp服务from aiohttp import web async def handle_index(request): return web.Response(textHello, aiohttp!) async def handle_greet(request): name request.match_info.get(name, world) return web.json_response({message: fHello, {name}}) app web.Application() app.router.add_get(/, handle_index) app.router.add_get(/greet/{name}, handle_greet) if __name__ __main__: web.run_app(app, host0.0.0.0, port8080)这段代码创建了一个监听8080端口的Web服务/返回纯文本/greet/{name}返回JSON。你可以直接在浏览器访问http://localhost:8080/greet/aiohttp验证。几个值得注意的点web.run_app()内部会启动事件循环所以不需要再额外调用asyncio.run()。request.match_info可以拿到URL路径里的命名参数这是写RESTful接口时最常用的功能之一。web.json_response()会自动设置Content-Type: application/json并序列化字典。5.2 读取请求体和查询参数实际接口肯定要接收参数。查询参数用request.queryasync def handle_search(request): keyword request.query.get(keyword, ) page int(request.query.get(page, 1)) return web.json_response({keyword: keyword, page: page})POST JSON请求体用await request.json()async def handle_create_user(request): try: data await request.json() except Exception: return web.json_response({error: invalid JSON}, status400) username data.get(username) if not username: return web.json_response({error: username is required}, status400) return web.json_response({status: ok, username: username}, status201)5.3 在服务器端发起客户端请求到这里aiohttp的威力才真正体现出来——一个服务端handler里可以同时发起多个外部HTTP请求而且不会阻塞其他handler的执行。比如做一个简单的聚合服务前端请求/aggregate后端同时去请求两个外部API汇总后返回。import aiohttp from aiohttp import web async def fetch_url(session, url): async with session.get(url) as resp: return await resp.text() async def handle_aggregate(request): urls [https://api.example.com/data/1, https://api.example.com/data/2] async with aiohttp.ClientSession() as session: results await asyncio.gather(*(fetch_url(session, u) for u in urls)) return web.json_response({count: len(results)})这个聚合服务在等待外部API响应的过程中事件循环可以继续处理其他请求所以并发能力比同步实现强很多。5.4 跨域、中间件和静态文件写前端页面时经常遇到跨域问题。aiohttp可以通过中间件来统一处理CORSasync def cors_middleware(app, handler): async def middleware_handler(request): response await handler(request) response.headers[Access-Control-Allow-Origin] * response.headers[Access-Control-Allow-Methods] GET, POST, PUT, DELETE, OPTIONS response.headers[Access-Control-Allow-Headers] Content-Type return response return middleware_handler app web.Application(middlewares[cors_middleware])静态文件服务用web.static()app.router.add_static(/static/, path./static, namestatic)5.5 按需关闭客户端Session服务器里要是有自己的ClientSession记得在应用退出时关闭。可以用on_cleanup信号app web.Application() async def on_startup(app): app[session] aiohttp.ClientSession() async def on_cleanup(app): await app[session].close() app.on_startup.append(on_startup) app.on_cleanup.append(on_cleanup)这个模式我强烈建议养成习惯否则每次重启服务都会报连接泄漏的警告。6. 爬虫场景联调并发抓取与数据回写写完客户端和服务端我觉得还是得用爬虫这个最常见的场景把两部分串起来。毕竟从热词搜索来看很多人学aiohttp就是为了加速爬虫。6.1 一个完整的并发爬虫骨架假设要抓取一个分页列表的所有详情页典型的流程是先抓列表页拿到详情页URL再并发抓取详情页最后把解析结果存到本地文件。import aiohttp import asyncio import json import time async def fetch_text(session, url): async with session.get(url) as resp: if resp.status ! 200: return None return await resp.text() async def crawl_list(session, list_url): 抓列表页返回详情页URL列表这里用xpath或者正则解析 html await fetch_text(session, list_url) if not html: return [] # 这里假设用正则提取出 /detail/xxx.html 形式的链接 import re detail_urls re.findall(r/detail/(\d)\.html, html) return [fhttps://example.com/detail/{u}.html for u in detail_urls] async def crawl_detail(session, detail_url): 抓详情页提取标题和正文这里简化为提取title html await fetch_text(session, detail_url) if not html: return None import re title_match re.search(rtitle(.*?)/title, html) title title_match.group(1) if title_match else unknown return {url: detail_url, title: title} async def main(): LIST_URL https://example.com/list/1 async with aiohttp.ClientSession() as session: detail_urls await crawl_list(session, LIST_URL) print(f获取到 {len(detail_urls)} 个详情页URL) # 给信号量限速避免请求太密集 semaphore asyncio.Semaphore(5) async def bounded_fetch(url): async with semaphore: return await crawl_detail(session, url) tasks [bounded_fetch(url) for url in detail_urls] results await asyncio.gather(*tasks, return_exceptionsTrue) valid_results [r for r in results if isinstance(r, dict)] with open(results.json, w, encodingutf-8) as f: json.dump(valid_results, f, ensure_asciiFalse, indent2) print(f成功抓取 {len(valid_results)} 条数据) asyncio.run(main())6.2 信号量限速别把目标服务器打挂上面代码里用了asyncio.Semaphore(5)这是一个非常重要的保护机制。异步框架的并发效率太高如果不限流一瞬间可能发出几百个请求轻则被封IP重则给对方服务器带来压力。Semaphore的原理是维护一个计数器获取信号量时计数器减一如果计数器为0后续协程会阻塞等待直到其他协程释放信号量。可以把信号量理解为“最多同时只有5个人在过闸机”。我这里设置的是5实际根据自己的网络情况和目标网站的承受能力调整。一般爬虫控制在5到20之间比较安全。6.3 重试机制和错误处理网络请求永远可能失败重试是必须的。用aiohttp写重试其实很简单因为协程可以被反复调用。这里我用一个装饰器来实现import asyncio async def retry_async(func, *args, retries3, delay1, **kwargs): for attempt in range(retries): try: return await func(*args, **kwargs) except Exception as e: if attempt retries - 1: raise await asyncio.sleep(delay * (attempt 1)) # 指数退避 return None调用的时候包一层async def bounded_retry(url): async with semaphore: return await retry_async(crawl_detail, session, url)重试之间的延迟建议用指数退避第一次等1秒第二次等2秒再往后等4秒。这种策略对服务器的压力更平滑。6.4 数据入库的方式抓下来的数据要写数据库可以用异步驱动比如asyncpgPostgreSQL或aiomysqlMySQL。如果不想引入额外依赖也可以像上面的例子一样先写到JSON文件再通过定时任务同步到数据库。对于数据量不大的场景写文件其实够了简单可靠。有一点值得注意不要在事件循环里做同步的文件写入或数据库查询这些操作同样会阻塞事件循环。如果数据库是同步的可以考虑用asyncio.to_thread把它丢到线程池里# 假设sync_save_to_db是同步函数 await asyncio.to_thread(sync_save_to_db, item)7. 避坑记录官方文档不会写的那几件事aiohttp整体设计得不错但深入用下来有一些坑是文档里不明显、实际运行中才会暴露的。我把这几年印象最深的几个问题和排查过程记录下来希望能帮你少走弯路。7.1 客户端Session和连接池泄漏这个坑我踩过不止一次。早期写服务时我在每个handler里都创建新的ClientSession用完后直接session.close()。听起来没问题但跑一段时间后会发现TCP连接数一直在涨。原因是ClientSession底层维护了连接池即使请求结束了连接也会保留一段时间以便复用。正确做法是全局只创建一个ClientSession整个进程共用。我的习惯是绑定到应用实例上在on_startup里创建、on_cleanup里关闭像上面5.5写的那样。如果你是在普通脚本里用也要注意asyncio.run(main())里的main()协程结束时async with aiohttp.ClientSession()会关闭session这个模式是安全的。但如果用的是较早的loop asyncio.get_event_loop()写法记得手动关闭session和事件循环。7.2 忘记await导致“coroutine was never awaited”警告刚转aiohttp时最常见的报错就是RuntimeWarning: coroutine xxx was never awaited。原因很简单写代码的时候漏了await。比如resp session.get(url) # 忘了await html resp.text() # 这里拿到的不是字符串而是coroutine对象这类错误比较隐蔽因为不一定会立刻报错只是运行时有警告并且变量值不是你预期的。排查方法也很直接看到警告就去检查哪一行少了await。还有一个容易混淆的地方resp.text()要awaitresp.json()要awaitresp.read()要await但resp.status、resp.headers不需要await。如果调用链里有嵌套的异步方法要确保每一层都写了await。7.3 事件循环Event loop is closed错误这个错误通常出现在多次调用asyncio.run()之后。最早的asyncio.get_event_loop()会在每次循环关闭后让某些资源失效解决办法是尽量不要在同一个线程里反复创建和关闭事件循环。更常见的是在Jupyter Notebook里用aiohttp。Notebook本身已经有一个正在运行的事件循环你再调用asyncio.run()会冲突。解决办法是使用await直接在Notebook的loop里执行异步代码或者使用nest_asyncio补丁import nest_asyncio nest_asyncio.apply()不过nest_asyncio这个库我不是每次都推荐它修改了asyncio的事件循环策略副作用未知最好仅在Notebook环境里使用。7.4 调用同步库阻塞事件循环这是个设计层面的坑不算bug但特别容易踩。举个例子你的handler里拿到了响应HTML然后调用了一个同步的第三方库去解析HTML或者做关键词匹配这个同步调用可能耗时几百毫秒。在同步编程里几百毫秒不算什么但在异步框架里这意味着整个事件循环卡住几百毫秒所有其他请求都得排队等。解决办法把CPU密集或同步IO的部分用asyncio.to_thread()丢给线程池执行from aiohttp import web import asyncio async def handle_parse(request): data await request.read() result await asyncio.to_thread(sync_heavy_parse, data) return web.json_response({result: result})asyncio.to_thread()在Python 3.9以上可用非常方便。如果你的Python版本较低可以用loop.run_in_executor(None, func, args)达到类似效果。7.5 不要在每个请求里重建正则或解析器这个是从性能优化角度给的一个提醒。异步框架的并发高如果每个请求进来都re.compile()一次正则、或者初始化一个解析器对象资源的浪费会被放大几百倍。把这类对象定义在模块级别或者应用实例上复用它import re TITLE_RE re.compile(rtitle(.*?)/title) async def handle(request): html await request.text() m TITLE_RE.search(html) ...7.6 优雅关闭连接池和后台任务Web应用里如果开了后台任务比如定时抓取数据的循环要确保应用关闭时这些任务也被取消或等待完成。aiohttp的on_cleanup钩子里可以做这件事async def on_cleanup(app): # 取消后台任务 for task in app.get(background_tasks, []): task.cancel() try: await task except asyncio.CancelledError: pass # 关闭ClientSession if session in app: await app[session].close()7.7 Python版本不同引起的行为差异最后提一个比较隐蔽的问题。async with session.get(url) as resp:这个写法在不同Python版本下对响应体释放时机的管理略有差异。在老版本中如果你没有把响应体读完就离开了async with块连接可能不会被正确放回连接池。所以如果发现有连接堆积检查一下是不是所有响应都被完整读取了。如果确实不需要读取体可以主动调用resp.release()。8. 进阶方向和数据分析、量化策略结合的思路从热搜词里能看到很多人也在搜“python数据分析与可视化”、“python量化交易策略代码”。aiohttp在这些领域同样能发挥作用——数据源和交易接口通常是HTTP API用异步框架可以显著提高数据获取效率。我自己的一个实践是用aiohttp写了一个小的行情数据采集服务每秒钟从几个公开API拉取行情快照写入内存中的环形缓冲区同时提供一个/latest接口给分析模块查询最新行情。整个服务用不到100行代码却能在单线程下轻松管理几十个WebSocket或HTTP长连接。对于个人级别的量化回测和研究来说这个方案足够轻量、足够快。如果你是非技术背景或者刚开始学Pythonaiohttp可以作为理解“异步编程”非常好的切入点。从curl到requests再到aiohttp你对IO模型的认知会逐步加深。写爬虫时会理解为什么“等待”是有代价的写Web服务时会理解为什么“同时服务很多请求”是很有价值的能力。我觉得整个学习路径可以这样安排先用requests写一个简单的同步爬虫跑通之后再用aiohttp重写一遍对比一下耗时和资源占用然后尝试用aiohttp写一个只返回JSON的本地小服务把爬到的数据通过接口暴露出来。走完这两步你基本就掌握aiohttp的核心用法了。9. 一些个人的经验之谈写了这么多最后分享一点自己的使用习惯。我现在的项目结构一般是一个crawler.py负责异步抓取数据一个server.py负责提供API两者共享一个storage.py模块做数据读写。在crawler.py里ClientSession作为模块级变量避免重复创建在server.py里ClientSession挂在app对象上通过request.app[session]访问。这样的结构职责分明也方便测试。另外调试异步代码时我会在关键位置加logger而不是print。aiohttp的日志如果不配置默认是一堆DEBUG级别的网络日志很吵。我习惯设置import logging logging.basicConfig(levellogging.WARNING)这样只显示警告和错误排查问题的时候再临时调到DEBUG。还有一个很实用的小技巧写客户端请求时给每个请求加一个request_idUUID通过HTTP头传给服务器配合日志能快速追踪一次完整的请求链路。在复杂场景下这个习惯能帮你节省大量排查时间。aiohttp的路还很远比如WebSocket支持、中间件机制、流式上传下载这些高级功能我这次都没有展开。如果你用到了WebSocket实时通信或者需要处理大文件上传aiohttp同样有对应的方案。到时候再写一篇单独聊。这里说的所有代码我建议你都亲手敲一遍。异步编程不是看会的是跑会的。你只有在自己的机器上亲眼看到并发请求的耗时变化、亲手排过“coroutine was never awaited”的错才算是真正入门了。