1. 项目概述这不是一本电子书而是一份我压箱底的异步编程实战手记“Python Async Programming: The Complete Guide”——这个标题听起来像某本畅销技术书的副标题但在我这儿它根本不是书而是过去三年里我带团队重构五个高并发服务时用掉的七本笔记本、三百二十七次失败重试、以及在凌晨三点盯着asyncio.run()报错堆栈时写下的血泪笔记。它不讲“什么是协程”不画抽象的状态机图只回答一个最实际的问题当你的API响应时间从200ms飙到2秒当数据库连接池天天爆满当你发现加机器根本不管用时async到底该怎么用、用在哪、为什么这么用才不翻车核心关键词是asyncio、协程、事件循环、异步IO、并发控制、错误传播它们不是概念而是你明天就要改的那行await db.fetch()背后的执行逻辑。适合谁适合已经写过requests.get()但一看到aiohttp.ClientSession()就头皮发麻的中级开发者适合被产品经理催着把报表导出接口从30秒压到3秒的后端工程师也适合刚学完async/await语法、却在真实项目里连asyncio.gather()和asyncio.create_task()该用哪个都拿不准的实践者。这不是理论课这是手术刀——我们只切开真实代码里的脓包不碰教科书上的标本。2. 内容整体设计与思路拆解为什么放弃“教科书式”异步教学2.1 从“语法正确”到“生产可用”的断层才是真正的痛点我见过太多人卡在第一步写了个async def hello(): return world跑通了就以为自己会异步了。结果上线后await asyncio.sleep(1)没改掉整个服务在高峰期直接挂掉——因为sleep(1)是同步阻塞它让整个事件循环停摆1秒所有其他请求全在排队。这暴露了一个致命问题绝大多数异步教程只教“怎么写”不教“怎么想”。它们把async当成一种新语法糖而不是一套全新的执行模型。我的设计思路彻底反向不从async/await开始而是从一次真实的线上故障复盘切入。比如去年我们一个订单查询服务在促销期间QPS从800突增到4500响应P95从180ms跳到2.3秒。监控显示CPU只有35%但asyncio事件循环的_run_once耗时暴涨。最终定位到三行代码# 错误示范看似异步实则同步黑洞 async def get_order_detail(order_id): # 这里调用了同步SDK它内部用requests.get()会阻塞整个event loop sync_result legacy_payment_sdk.check_status(order_id) await db.update_order_status(order_id, sync_result) return sync_result你看async def声明了函数await也写了但它调用的legacy_payment_sdk.check_status()是纯同步的。这个函数一执行事件循环就卡死后面所有协程全得等它。所以我的完整指南第一原则就是async不是魔法它只对真正异步的IO操作生效对同步代码它连个屁用都没有。整个内容骨架围绕“识别阻塞点→替换为异步实现→控制并发粒度→处理错误传播”这条真实排障链路展开每一步都对应一个线上踩过的坑。2.2 工具选型为什么只聚焦asyncio原生库拒绝trio或curio有人问为什么不讲trio它API更简洁错误处理更友好。我的答案很直接你在公司代码库里能说服CTO把生产环境的asyncio换成trio吗asyncio是Python标准库从3.7起就是async/await的官方载体Django 4.1、FastAPI、Starlette、aiohttp、aiomysql……所有主流异步框架都基于它构建。trio再优雅它也是第三方库意味着额外的维护成本、兼容性风险、以及团队学习曲线。我试过在内部技术分享中引入trio结果运维同事第一句话是“它和我们现有的Prometheus监控集成文档在哪”——没有。而asyncio的loop.set_debug(True)、asyncio.all_tasks()、asyncio.current_task()这些调试工具文档齐全社区案例多出了问题搜Stack Overflow就能找到答案。所以本指南所有代码、所有示例、所有排查命令全部锁定asyncio原生API。这不是保守而是对生产环境的敬畏。你不需要成为asyncio源码贡献者但必须吃透BaseEventLoop的三个核心方法run_until_complete()、run_forever()、create_task()。它们是你掌控异步世界的三把钥匙。2.3 场景优先为什么把“数据库操作”放在第一章而不是“协程原理”因为90%的异步需求都来自IO瓶颈。而IO里数据库是头号杀手。你写个SELECT * FROM orders WHERE user_id ?如果用同步驱动它会发起TCP连接、发送SQL、等待网络往返、解析结果集——这一整套流程在同步模式下线程就卡在那里干等。换成异步驱动如aiomysql、asyncpg它把连接、查询、结果获取全部注册到事件循环里线程可以去处理其他请求。所以我的结构完全按高频痛点场景排列数据库 → HTTP请求 → 文件IO → 第三方SDK封装 → 并发控制 → 错误处理 → 调试监控。每一章开头都是一段真实日志截图已脱敏“2023-08-15 14:22:07 ERROR [db] Connection pool exhausted: maxsize10, overflow20”。然后才展开为什么池子会爆maxsize和overflow怎么设aiomysql.create_pool()的minsize参数到底影响什么这种结构让你学完立刻能回去改代码而不是学完一堆概念面对报错还是两眼一抹黑。3. 核心细节解析与实操要点async/await不是语法糖是执行模型的切换开关3.1 协程对象coroutine object的本质它不是函数而是一个状态机实例很多人以为async def func()定义的是一个函数await func()是调用它。大错特错。async def定义的是一个协程工厂函数。当你调用它时比如coro fetch_data()返回的不是结果而是一个协程对象coroutine object。这个对象本身不执行任何代码它只是保存了函数体、局部变量、当前执行位置类似生成器的gi_frame。真正的执行始于await coro或asyncio.run(coro)。你可以把它理解成一个待签发的“任务单”上面写着“去查数据库拿到结果后通知我”。await的作用就是把这张单子交给事件循环由循环调度执行。验证这一点很简单import asyncio async def demo(): print(Step 1) await asyncio.sleep(0.1) print(Step 2) return done # 注意这里只是创建协程对象不打印任何东西 coro demo() print(type(coro)) # class coroutine print(coro.cr_running) # False还没运行 # 真正执行才会触发print result asyncio.run(coro) # 输出 Step 1, Step 2, 返回 done这个细节至关重要。它解释了为什么你不能在普通函数里直接await因为普通函数没有事件循环上下文await找不到“收单人”。这也是RuntimeError: no running event loop报错的根源。解决方案只有两个要么用asyncio.run()启动一个新循环仅限脚本顶层要么确保你在另一个async函数里await。我见过最离谱的错误是有人在Django视图里写# Django视图同步环境 def order_view(request): # 错误这里没有event loopawait会直接报错 data await fetch_order_async(request.GET[id]) return JsonResponse(data)正确做法是用支持异步的框架如FastAPI或Django 4.1的async def视图。记住await不是暂停是交权async def不是函数是任务单生成器。3.2 事件循环Event LoopPython异步的“交通指挥中心”如果你把Python进程比作一座城市那么事件循环就是它的交通指挥中心。它不自己开车不执行业务逻辑但它决定哪辆车协程在哪个路口IO事件通行。asyncio默认使用ProactorEventLoopWindows或SelectorEventLoopLinux/macOS。后者基于epollLinux或kqueuemacOS能高效监听成千上万个文件描述符socket、pipe等的读写就绪状态。当一个协程执行await aiohttp.get(url)时它实际做了三件事向事件循环注册一个“当这个socket可读时请唤醒我”的回调主动让出控制权yield事件循环得以调度下一个协程当网络数据到达epoll通知事件循环循环立即调用之前注册的回调恢复该协程的执行。这就是“单线程高并发”的秘密CPU不浪费在空等IO上而是在多个IO等待任务间快速切换。但这也带来一个硬约束所有耗时的CPU计算都必须被切碎否则会阻塞整个循环。比如你写个await cpu_intensive_task()而这个任务内部是个百万级循环那所有其他协程就得等它算完。解决方案是loop.run_in_executor()把CPU密集型任务扔给线程池或进程池执行避免阻塞主线程。我在处理图片缩略图生成时就吃过亏一开始直接在协程里用PIL做resize结果上传接口P99飙升到8秒。改成import asyncio from concurrent.futures import ProcessPoolExecutor # CPU密集型任务丢给进程池 def resize_image_sync(image_path, size): from PIL import Image img Image.open(image_path) return img.resize(size) async def resize_image_async(image_path, size): loop asyncio.get_running_loop() # 在进程池中执行不阻塞event loop result await loop.run_in_executor( ProcessPoolExecutor(max_workers4), resize_image_sync, image_path, size ) return resultmax_workers4不是拍脑袋定的。我实测过在16核服务器上max_workers设为CPU核心数的1.5倍即24时图片处理吞吐量最高但设为32时进程切换开销反而增大吞吐下降12%。这个数字必须根据你的具体硬件和任务类型压测得出。3.3 异步IO的“真·异步”门槛驱动、协议、操作系统三者缺一不可很多开发者以为只要把requests换成aiohttp代码就自动异步了。错。aiohttp只是提供了异步HTTP客户端但它的底层依赖是否真异步才是关键。我们来拆解一次aiohttp.get()的调用链aiohttp.ClientSession.get()→ 创建一个ClientRequest对象ClientRequest.send()→ 调用self._writer.write_headers()_writer是一个StreamWriter它最终调用self._transport.write()self._transport是_SelectorSocketTransportLinux下它包装了一个socket对象这个socket必须是非阻塞模式socket.setblocking(False)且其文件描述符fd被注册到epoll实例中。看出来了吗“异步”不是某个库的魔法而是从应用层aiohttp→传输层asyncio transport→系统调用epoll_wait的全链路协作。如果你用的数据库驱动是pymysql同步那await db.execute()永远是假异步必须换aiomysql或asyncpg。同理文件IOopen()是同步的await它会报错必须用aiofiles库它内部用loop.run_in_executor()把os.read()扔给线程池。我曾经为了优化日志写入尝试用asyncio.to_thread()Python 3.9包装logging.info()结果发现性能反而下降15%——因为日志写入本身很快线程切换开销超过了收益。最后改用内存缓冲定时刷盘这才是正解。判断一个操作是否真异步只有一个标准它是否最终调用了epoll_ctl()或kqueue()系统调用并将fd注册为非阻塞。其他都是障眼法。4. 实操过程与核心环节实现从零搭建一个抗住5000 QPS的异步订单服务4.1 环境准备与依赖锁定为什么pyproject.toml比requirements.txt更可靠新手常犯的错误是直接pip install aiohttp asyncpg然后发现asyncpg在Python 3.12下编译失败或者aiohttp的某个版本和asyncio的TaskGroup3.11有兼容问题。生产环境必须杜绝这种不确定性。我的方案是强制使用pyproject.toml并精确锁定所有依赖的次要版本[build-system] requires [setuptools45, wheel, setuptools_scm[toml]6.2] build-backend setuptools.build_meta [project] name async-order-service version 1.0.0 dependencies [ aiohttp3.9.5, # 不用~用避免小版本升级引入breaking change asyncpg0.29.0, # asyncpg 0.28.x在高并发下有连接泄漏bug pydantic2.7.1, # FastAPI依赖版本必须匹配 uvloop0.19.0, # 可选但强烈推荐比默认loop快30% ] [project.optional-dependencies] dev [pytest-asyncio0.23.5, mypy1.10.0]uvloop是asyncio的Cython加速版它用libuvNode.js的事件循环重写了底层实测在我们的订单服务中QPS从3200提升到4100P95延迟降低22%。安装它只需一行pip install uvloop然后在应用入口加import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # 必须在asyncio.run()之前调用注意uvloop不支持Windows所以开发机用Mac/LinuxCI/CD流水线用Linux容器保持环境一致。pyproject.toml还解决了另一个痛点aiohttp的ClientSession需要手动关闭否则连接不释放。我们用async with确保资源清理# 正确自动管理session生命周期 async def fetch_product_info(product_id: str) - dict: async with aiohttp.ClientSession() as session: async with session.get(fhttps://api.example.com/products/{product_id}) as resp: return await resp.json() # 错误session可能未关闭导致连接泄漏 # session aiohttp.ClientSession() # resp await session.get(...) # await session.close() # 忘关完蛋。4.2 数据库层asyncpg连接池的12个关键参数详解asyncpg是目前最快的PostgreSQL异步驱动但它的连接池配置极其精细。我花了两周时间压测总结出生产环境必须关注的12个参数asyncpg.create_pool()的kwargs参数名推荐值为什么这么设实测影响min_size10保证池中总有10个空闲连接避免冷启动时创建连接的延迟P95延迟降低150msmax_size50单个pool最大连接数超过此数新请求会排队防止DB被打爆max_inactive_connection_lifetime300.0 (5分钟)连接空闲超5分钟自动回收防长连接泄漏减少DB端TIME_WAIT连接数40%max_queries50000单个连接执行5万次查询后强制回收防内存泄漏避免asyncpg内部query cache膨胀command_timeout10.0单条SQL执行超10秒强制中断防慢查询拖垮整个池防止一个慢SQL阻塞所有请求server_settings{application_name: order-service}给DB打标签便于PG Admin里区分来源运维排查必备initlambda conn: conn.execute(SET timezone UTC)每个新连接初始化时执行避免时区混乱其他参数如loop已废弃、ssl必须设为require、host/port/database/user/password是基础项。最关键的min_size和max_size我通过pg_stat_activity视图实时监控-- 查看当前所有连接及其状态 SELECT application_name, state, COUNT(*) FROM pg_stat_activity WHERE backend_type client backend GROUP BY application_name, state;当state idle in transaction的连接数持续5说明有事务没提交要查代码当state active的连接数接近max_size说明DB处理不过来得优化SQL或加索引。连接池不是越大越好。我们曾把max_size设为200结果DB CPU飙到95%因为PostgreSQL的每个连接都要消耗内存和锁资源。最终平衡点是50配合DB的max_connections200留出余量给其他服务。4.3 HTTP客户端aiohttp的ClientSession高级用法与陷阱aiohttp.ClientSession是异步HTTP的基石但它的默认行为充满陷阱。第一个坑DNS缓存。默认情况下aiohttp会缓存DNS解析结果10分钟resolve_cache_duration300但如果目标服务做了蓝绿发布IP变了你的请求会持续失败10分钟。解决方案是禁用缓存或设极短时间import aiohttp from aiohttp import TCPConnector connector TCPConnector( limit100, # 同时最多100个连接 limit_per_host30, # 对同一host最多30个连接防被封 keepalive_timeout30, # 连接保活30秒 use_dns_cacheFalse, # 关键禁用DNS缓存 ttl_dns_cache10 # 如果要用缓存设为10秒 ) async with aiohttp.ClientSession(connectorconnector) as session: # 所有请求共享此connector pass第二个坑SSL验证。测试环境常用自签名证书aiohttp默认严格校验会报aiohttp.client_exceptions.ClientConnectorCertificateError。生产环境绝不能关但测试环境可以# 测试环境临时方案仅限localhost connector TCPConnector(sslFalse) # 关闭SSL验证 # 生产环境必须用真实证书 connector TCPConnector(sslTrue) # 默认就是True第三个坑超时设置。timeout参数有四个维度必须全设timeout aiohttp.ClientTimeout( total30, # 整个请求总超时含DNS、连接、读取 connect10, # 连接建立超时DNSTCP握手 sock_read15, # socket读取超时从收到第一个字节到结束 sock_connect10 # socket连接超时TCP握手完成 ) async with aiohttp.ClientSession(timeouttimeout) as session: pass漏设sock_read遇到对方服务器卡住不发数据你的协程就永远挂起。我在线上遇到过一次支付网关偶发卡顿sock_read没设导致整个订单服务的连接池被占满雪崩。超时不是可选项是安全阀。最后aiohttp的raise_for_status()必须显式调用否则HTTP 4xx/5xx错误不会抛异常你会得到一个ClientResponse对象里面status500但代码继续往下走造成数据不一致。4.4 并发控制asyncio.Semaphore与asyncio.BoundedSemaphore的生死抉择高并发下你不能让1000个协程同时去调用同一个外部API否则会被限流拉黑。asyncio.Semaphore是你的流量闸门。但选Semaphore还是BoundedSemaphore区别在于Semaphore的计数可以被release()多次超过初始值BoundedSemaphore则严格限制release()超过初始值会抛ValueError。生产环境必须用BoundedSemaphore因为它能防止因代码bug导致的信号量泄露。# 全局信号量限制对支付网关的并发调用不超过5 PAYMENT_SEM asyncio.BoundedSemaphore(5) async def call_payment_gateway(order_id: str) - bool: async with PAYMENT_SEM: # 自动acquire/release async with aiohttp.ClientSession() as session: async with session.post( https://pay.example.com/charge, json{order_id: order_id} ) as resp: if resp.status 200: return True raise Exception(fPayment failed: {resp.status})但信号量只是粗粒度控制。更精细的是动态并发控制。比如我们有个商品搜索接口要并发查ES、查MySQL、查Redis。但ES查询快50msMySQL慢300msRedis最快5ms。如果用统一信号量会拖慢整体。我的方案是分层# 三层信号量独立控制 ES_SEM asyncio.BoundedSemaphore(20) # ES扛得住高并发 DB_SEM asyncio.BoundedSemaphore(5) # MySQL连接池小限制严 REDIS_SEM asyncio.BoundedSemaphore(100) # Redis轻松扛 async def search_products(keyword: str): # 并发发起三个请求但各自受不同信号量限制 es_task asyncio.create_task(_search_es(keyword)) db_task asyncio.create_task(_search_db(keyword)) redis_task asyncio.create_task(_search_redis(keyword)) results await asyncio.gather(es_task, db_task, redis_task, return_exceptionsTrue) # 处理results...asyncio.gather()的return_exceptionsTrue是关键。它确保即使一个任务抛异常其他任务仍会继续执行并返回结果异常对象而不是整个gather失败。这比asyncio.wait()更实用因为你通常需要尽可能多地拿到数据哪怕部分失败。4.5 错误处理与传播为什么try/except在协程里会失效这是最隐蔽的坑。看这段代码async def bad_error_handling(): try: await risky_operation() # 可能抛AsyncIOError except AsyncIOError as e: log.error(Caught AsyncIOError, exc_infoe) # 这里处理... return None except Exception as e: log.error(Unexpected error, exc_infoe) return None # 调用它 result await bad_error_handling()看起来完美错。risky_operation()如果是一个async def函数它内部抛出的异常会被await捕获并向上抛try/except能抓到。但如果risky_operation()是同步函数比如time.sleep(10)它抛的是KeyboardInterruptawait无法捕获异常会直接冒泡到事件循环导致整个程序崩溃。更糟的是asyncio.gather()默认行为是“任一失败则全部失败”这在微服务调用中是灾难性的。正确姿势是所有外部调用必须包裹在asyncio.wait_for()中并指定超时所有并发任务必须用return_exceptionsTrue。同时定义清晰的错误分类class OrderServiceError(Exception): 所有订单服务自定义异常基类 class PaymentGatewayError(OrderServiceError): 支付网关调用失败 class DatabaseError(OrderServiceError): 数据库操作失败 async def safe_call_payment(order_id: str) - bool: try: # 加超时防死等 return await asyncio.wait_for( _call_payment_internal(order_id), timeout8.0 # 支付网关SLA是5秒这里留3秒buffer ) except asyncio.TimeoutError: log.warning(fPayment timeout for order {order_id}) raise PaymentGatewayError(fTimeout calling payment for {order_id}) except aiohttp.ClientError as e: log.error(fHTTP error calling payment: {e}) raise PaymentGatewayError(fHTTP error: {e}) except Exception as e: log.exception(fUnexpected error in payment call: {e}) raise PaymentGatewayError(fUnexpected: {e}) # 在主流程中 results await asyncio.gather( safe_call_payment(order1), safe_call_payment(order2), return_exceptionsTrue ) for i, res in enumerate(results): if isinstance(res, Exception): log.error(fOrder {i1} failed: {res}) # 记录失败订单后续补偿 else: log.info(fOrder {i1} succeeded: {res})这样错误被精准分类、记录、隔离不会因为一个订单失败导致整个批次处理中断。5. 常见问题与排查技巧实录那些让我凌晨三点爬起来的报错5.1 “RuntimeWarning: coroutine ‘xxx’ was never awaited” —— 协程对象的“幽灵泄漏”这个警告不是报错但比报错更可怕。它意味着你创建了一个协程对象却从未await它也从未把它交给事件循环。这个对象会一直存在占用内存直到被垃圾回收。在长周期服务中它会导致内存缓慢增长最终OOM。典型场景在if语句里忘了awaitif should_fetch: fetch_data() # 错这里应该 await fetch_data()把await写在了条件表达式里# 错这行代码根本不会执行因为await只能在async函数里 result await fetch_data() if condition else None调用async函数时漏掉了括号变成函数对象而非调用# 错fetch_data是函数对象不是协程对象 coro fetch_data # 应该是 fetch_data()排查技巧在pyproject.toml里加[tool.pytest.ini_options]启用asyncio_mode auto并用pytest-asyncio插件。它会在测试时自动检测未await的协程。更重要的是在应用入口加全局钩子import asyncio import warnings # 捕获所有未await的协程警告 def warn_on_unclosed_coro(coro): warnings.warn(fUnclosed coroutine: {coro}, RuntimeWarning) # Python 3.12 可以用 asyncio.on_cancelled() # 旧版本用这个hack original_create_task asyncio.create_task def patched_create_task(coro, *, nameNone): if not asyncio.iscoroutine(coro): raise TypeError(fexpect a coroutine, got {type(coro)}) task original_create_task(coro, namename) # 任务完成时检查是否异常 def done_callback(t): if t.cancelled(): pass elif t.exception() is not None: log.error(Task failed, exc_infot.exception()) task.add_done_callback(done_callback) return task asyncio.create_task patched_create_task5.2 “Event loop is closed” —— 事件循环的“猝死”之谜这个错误通常出现在应用退出时或者在asyncio.run()之后又试图创建新任务。根本原因是asyncio.run()会创建一个新事件循环执行完协程后自动关闭该循环。如果你在run()之后又调用asyncio.create_task()就会报错因为循环已关。典型场景在asyncio.run(main())之后还有代码试图await别的东西在atexit钩子里调用异步清理函数使用threading.Thread在子线程里调用asyncio.run()但主线程已退出。根治方案永远不要在asyncio.run()之外的地方假设事件循环存在。所有异步代码必须包裹在async def main()里由asyncio.run(main())统一启动。清理工作用asyncio.CancelledError捕获async def main(): # 启动服务 server await start_server() try: # 主循环 await server.serve_forever() except asyncio.CancelledError: # 优雅退出 log.info(Shutting down...) await server.close() await cleanup_resources() raise # 重新抛出让run()正常退出 if __name__ __main__: asyncio.run(main()) # 全局唯一入口5.3 “Task was destroyed but it is pending!” —— 任务的“未了心愿”这个错误意味着一个asyncio.Task对象被垃圾回收了但它还在事件循环里排队等着执行。常见于在asyncio.run()结束前你创建了一个asyncio.create_task()但没等它完成run()就结束了循环关闭任务被强制销毁。典型场景在main()里create_task()一个后台任务如心跳上报但main()很快结束asyncio.gather()里某个任务执行时间很长gather超时被取消但任务本身还在跑。解决方案所有后台任务必须显式await或asyncio.wait_for()。对于必须长期运行的后台任务用asyncio.create_task()并存储引用确保它不被GC# 全局存储后台任务防止被GC _background_tasks set() async def background_heartbeat(): while True: try: await send_heartbeat() except Exception as e: log.error(Heartbeat failed, exc_infoe) await asyncio.sleep(30) # 启动后台任务 task asyncio.create_task(background_heartbeat()) _background_tasks.add(task) # 任务完成后从集合中移除 task.add_done_callback(_background_tasks.discard)5.4 性能瓶颈定位asyncio的“透视眼”调试工具当服务变慢别急着加机器。先用asyncio自带的透视工具开启调试模式asyncio.run(coro, debugTrue)。它会检测长时间运行的协程100ms并警告检测未关闭的ClientSession、ConnectionPool显示事件循环的统计信息。查看所有活跃任务asyncio.all_tasks()。在出问题时加个调试端点from aiohttp import web async def debug_tasks(request): tasks asyncio.all_tasks() return web.json_response({ total_tasks: len(tasks), running_tasks: [t.get_coro().__name__ for t in tasks if not t.done()], pending_tasks: [t.get_coro().__name__ for t in tasks if t.done() and not t.cancelled()] }) app.router.add_get(/debug/tasks, debug_tasks)火焰图分析用py-spyRust写的无侵入式采样器pip install py-spy py-spy record -p pid -o profile.svg --duration 30生成的SVG图能清晰看到CPU时间花在_run_once、select、还是你的业务函数上。我用py-spy发现过一个经典问题asyncpg的fetchrow()在处理大JSON字段时内部json.loads()是同步的占了CPU 40%时间。解决方案是用asyncpg的record[json_field]直接获取字符串然后用asyncio.to_thread(json.loads, ...)在后台线程解析CPU占用降到5%。5.5 日志与监控如何让异步代码的log不再“乱序”异步环境下print()和logging.info()的输出顺序会乱。因为多个协程交替执行log语句穿插其中。解决方案是为每个请求分配唯一trace_id并在所有log中带上它。用contextvarsPython 3.7import contextvars import logging # 创建上下文变量 request_id_var contextvars.ContextVar(request_id, defaultunknown) # 自定义