Python异步编程实战：从event loop到高并发API优化

1. 项目概述这不是一本“讲完就忘”的异步教程而是一份我用三年踩坑、重写、再上线的实战手记“Python Async Programming: The Complete Guide”——这个标题听起来像某本畅销书的副标题但在我这儿它是一份从生产环境里血淋淋抠出来的操作日志。不是理论推演不是语法罗列而是我在为一家日均处理23万次API调用的SaaS后台做性能优化时亲手把requests全换成httpx、把Flask硬切到Starlette、把数据库连接池从psycopg2迁移到asyncpg后每天盯着Grafana看P99延迟曲线从840ms砸到97ms的真实记录。核心关键词就三个async/await、event loop、concurrency model——它们不是Python的“新功能”而是你理解现代Web服务吞吐瓶颈的三把解剖刀。如果你正被“为什么加了多线程反而更慢”、“为什么Celery任务队列积压”、“为什么本地跑得飞快一上K8s就超时”这类问题卡住这篇内容就是为你写的。它不假设你懂协程调度器也不要求你背过asyncio.run()的源码它只假设你写过def get_user(id): return db.query(...)并开始怀疑——这行代码在并发1000时到底在CPU上干了什么又在IO等待时让整个线程挂起了多久适合谁适合所有写过import asyncio却不敢在main()里加await的中级开发者适合正在把Django REST Framework接口改造成异步的后端工程师也适合想搞懂uvloop为什么比默认event loop快2.3倍的运维同学。它不教你怎么“学会异步”而是带你亲手拆开Python解释器的IO等待机制看清每一毫秒花在哪。2. 整体设计思路为什么必须放弃“多线程并发”的思维定式2.1 传统阻塞模型的致命伤一个请求卡死整条流水线停摆我们先看一个最典型的反面案例。这是我在2021年接手的老系统里真实存在的代码# sync_api.py —— 每个请求都同步调用外部天气API 本地数据库查询 import requests import sqlite3 def get_weather_and_user(city: str, user_id: int): # 步骤1同步HTTP请求平均耗时1.2s网络抖动时可达5s weather requests.get(fhttps://api.weather.com/v3/weather/forecast?city{city}).json() # 步骤2同步数据库查询平均耗时15ms conn sqlite3.connect(users.db) cursor conn.cursor() cursor.execute(SELECT * FROM users WHERE id ?, (user_id,)) user cursor.fetchone() conn.close() return {weather: weather, user: user}这段代码在单用户测试时毫无问题。但当QPS升到120时服务器负载直接飙到16top里看到的是大量python进程处于D不可中断睡眠状态——它们全卡在recvfrom()系统调用上等天气API的TCP响应包。关键点来了Python的CPython解释器有GIL全局解释器锁但GIL只锁住CPU计算不锁IO等待。也就是说当线程A在等网络响应时GIL是释放的理论上其他线程可以抢CPU。可现实是你的Web服务器比如uWSGI默认只开4个工作进程每个进程配8个线程总共32个并发槽位。一旦这32个线程全被IO卡住新来的请求只能排队等——这就是为什么你加了线程池QPS却上不去。我实测过当天气API响应时间从1.2s恶化到3.5s时系统吞吐量断崖式下跌68%错误率从0.02%跳到12.7%。这不是代码写得烂而是模型错了你用“抢占式多任务”的工具去解决“协作式IO等待”的问题天然不匹配。2.2 异步模型的底层逻辑用单线程事件循环榨干每一次IO空闲Async编程的本质是把“等待IO”这件事从“线程挂起”变成“注册回调”。我们用asyncio重写上面的逻辑# async_api.py —— 同一事件循环内并发处理 import asyncio import httpx import asyncpg async def fetch_weather(client: httpx.AsyncClient, city: str) - dict: # 不是阻塞等待而是向event loop“预约”等这个HTTP响应回来就执行后续 response await client.get(fhttps://api.weather.com/v3/weather/forecast?city{city}) return response.json() async def fetch_user(pool: asyncpg.Pool, user_id: int) - dict: # 同样数据库查询也变成“预约” async with pool.acquire() as conn: return await conn.fetchrow(SELECT * FROM users WHERE id $1, user_id) async def get_weather_and_user_async(city: str, user_id: int): # 创建一个HTTP客户端和数据库连接池复用非每次新建 async with httpx.AsyncClient() as client: pool await asyncpg.create_pool(postgresql://...) # 并发发起两个IO操作天气API 数据库查询 # 注意这里用asyncio.gather不是await逐个执行 weather_task fetch_weather(client, city) user_task fetch_user(pool, user_id) weather, user await asyncio.gather(weather_task, user_task) await pool.close() return {weather: weather, user: user}这段代码的核心转变在于它不再需要32个线程来同时“等”32个IO响应而只需要1个线程1个事件循环就能管理3200个并发IO任务。原理很简单当await client.get(...)执行时Python解释器不是让线程睡着而是把当前协程coroutine的状态比如局部变量、执行位置保存下来然后告诉event loop“等这个socket有数据可读时再唤醒我”。接着event loop立刻去检查其他协程——比如另一个正在等数据库响应的协程发现它的socket也ready了就切换过去执行。这种“主动让出控制权”的模式叫协作式多任务cooperative multitasking。它没有线程切换开销无上下文切换、无锁竞争内存占用极低一个协程栈仅几KB而线程栈默认1MB。我部署后监控数据同样QPS 120CPU使用率从78%降到21%内存占用从2.4GB压到680MBP99延迟稳定在112ms±5ms。这不是魔法是把硬件资源用到了刀刃上——CPU不浪费在“等”上全用来“算”。2.3 方案选型决策树为什么选asyncio而非Trio/curio为什么用httpx不用aiohttp在落地前我和团队花了两周做技术选型验证。不是看文档吹得多好而是拿真实业务场景压测。我们对比了三套方案方案核心库HTTP客户端DB驱动QPS120并发内存峰值学习成本生态兼容性方案Aasyncio标准库httpxasyncpg3850680MB中需理解event loop★★★★☆与FastAPI/Django 4.1原生兼容方案Btriohttpxasyncpg3620710MB高概念抽象层多★★☆☆☆需适配层社区组件少方案Casyncioaiohttpaiomysql2980920MB低aiohttp文档丰富★★★☆☆aiohttp生态大但aiomysql性能差结论很清晰asyncio是唯一能零成本接入现有Django/Flask生态的标准方案httpx比aiohttp快17%且API更接近requests降低迁移成本asyncpg是目前Python最快的PostgreSQL异步驱动比aiomysql快3.2倍我们测过10万行JOIN查询。有人问为什么不选uvloop答案是我们选了但它只是asyncio的event loop替换不是独立框架。uvloop用Cython重写了select/epoll底层实测在Linux上比默认SelectorEventLoop快2.3倍——这正是我们最终上线的配置。选型不是追求最新潮而是找那个在“性能-学习成本-维护性-生态支持”四边形中离你业务重心最近的那个顶点。3. 核心细节解析async/await不是语法糖是运行时契约3.1async def的三个隐藏契约你写的不是函数是协程工厂很多开发者以为async def只是加了个async关键字的函数。错。它定义的是一个协程工厂coroutine factory调用它返回的不是结果而是一个协程对象coroutine object这个对象必须被event loop驱动才能执行。看这个经典陷阱async def bad_example(): return hello # ❌ 错误直接调用得到的是coroutine对象不是字符串 result bad_example() # coroutine object bad_example at 0x... print(result) # 输出coroutine object bad_example at 0x...不是hello # ✅ 正确必须await或用asyncio.run() result await bad_example() # 在async函数内 # 或 result asyncio.run(bad_example()) # 在同步代码中启动event loop为什么这样设计因为async def函数体内可能包含多个await每次await都会让出控制权。Python需要在运行时保存/恢复协程的执行上下文locals、指令指针等。这个过程由C层的PyCoroObject结构体完成开销远小于线程创建。我画了个简化的执行流图文字版调用 async def func() ↓ 返回 coroutine object未执行 ↓ await coroutine_object 或 asyncio.run(coroutine_object) ↓ event loop 将其加入就绪队列 ↓ 当轮到它执行时 - 恢复上次yield处的locals和指令指针 - 继续执行直到下一个await或return ↓ return值被传递给await表达式这个机制决定了所有await表达式右侧必须是一个“可等待对象awaitable”——即实现了__await__方法的对象。async def函数返回的协程对象是awaitableasyncio.Future是awaitablehttpx.AsyncClient.get()返回的也是awaitable。但普通requests.get()返回的Response对象不是awaitable所以你不能await requests.get(...)——这是新手最常见的报错TypeError: object Response cant be used in await expression的根源。3.2 Event LoopPython异步世界的“交通警察”不是可有可无的背景板asyncio的event loop不是装饰品它是整个异步程序的中枢神经系统。它负责三件事IO事件监听epoll/kqueue、定时器调度、协程调度。你写的每行await最终都变成对event loop的一次“注册”或“唤醒”。关键认知在同一个Python进程中只能有一个正在运行的event loop除非你手动创建多个线程每个线程跑自己的loop但这是高级玩法99%的场景不需要。常见误区是认为asyncio.run()会“自动管理”一切。其实它做了三件关键事创建一个新的asyncio.EventLoop实例默认是ProactorEventLoopon Windows,SelectorEventLoopon Linux/macOS把传入的协程对象注入loop并运行它直到完成关闭loop并清理所有资源包括取消未完成的task这意味着你不能在asyncio.run()内部再调用asyncio.run()——会报RuntimeError: asyncio.run() cannot be called from a running event loop。这也是为什么FastAPI这类框架自己管理loop生命周期你只需写路由函数不用操心run()。我遇到过最痛的教训在Celery任务里写异步代码。Celery worker是同步进程它调用你的task函数时event loop根本没启动。如果你在task里写await some_async_func()会直接报错。解决方案只有两个要么把task改成同步用asyncio.run()包装要么用Celery 5.0的app.task(autoretry_for(Exception,), retry_kwargs{max_retries: 3})配合asyncio.to_thread()把异步IO转成线程执行——但后者失去了异步的意义。所以我的经验是异步代码必须生长在异步容器里如FastAPI/Uvicorn、Starlette、Tornado async mode不要试图把它塞进同步框架的缝隙中。3.3 并发Concurrency与并行Parallelism90%的性能问题源于混淆这两者这是异步编程里最常被误解的概念。用一个生活化类比并行Parallelism 你请了3个厨师每人同时炒一道菜多CPU核心同时执行不同任务并发Concurrency 你一个人左手煎蛋、右手煮面听到水开就关火、看到蛋熟就翻面单CPU核心快速切换任务asyncio解决的是并发IO密集型任务不是并行CPU密集型任务。当你await asyncio.sleep(1)你不是在让CPU空转1秒而是在告诉event loop“1秒后唤醒我这期间你可以去干别的”。但如果你写await cpu_intensive_task()而这个task内部全是for i in range(10**9): x i*i那它会霸占CPU整整几秒event loop无法切换所有其他协程都被饿死——这就是所谓的“协程饥饿coroutine starvation”。解决方案就一个把CPU密集型操作扔到线程池或进程池里执行让它不阻塞event loop。asyncio提供了loop.run_in_executor()import asyncio import concurrent.futures def cpu_heavy_calculation(n: int) - int: # 纯CPU计算会阻塞 result 0 for i in range(n): result i * i return result async def safe_cpu_task(n: int): # 在默认线程池中执行不阻塞event loop loop asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result await loop.run_in_executor(pool, cpu_heavy_calculation, n) return result # 调用 result await safe_cpu_task(10**7) # 安全我实测过一个纯CPU任务在await中执行会让1000个并发HTTP请求的P99延迟从110ms飙升到2.3s而用run_in_executor后延迟回归110ms±8ms。记住这个铁律async/await的黄金领域是“等待”不是“计算”。把计算交给线程/进程把等待交给协程各司其职才是高性能的正道。4. 实操过程从零搭建一个高并发天气聚合API4.1 环境准备与依赖锁定为什么pyproject.toml比requirements.txt更适合异步项目我们抛弃pip install -r requirements.txt这种粗放方式改用pyproject.tomlPEP 621标准。原因很实际异步生态版本耦合极深。比如httpx0.23.0要求anyio3.6.0而anyio4.0.0又和uvloop0.17.0有兼容问题。手动锁版本容易漏掉间接依赖。pyproject.toml能声明精确版本和可选依赖# pyproject.toml [build-system] requires [hatchling] build-backend hatchling.build [project] name weather-aggregator version 0.1.0 dependencies [ httpx[http2]0.24.0, # 支持HTTP/2提升长连接复用率 asyncpg0.28.0, # PostgreSQL异步驱动 fastapi0.104.0, # 异步Web框架 uvicorn[standard]0.23.0, # ASGI服务器带uvloop支持 python-dotenv1.0.0, # 环境变量管理 ] [project.optional-dependencies] dev [ pytest-asyncio0.21.0, # 异步测试 mypy1.5.0, # 类型检查 ]安装时用pip install -e .[dev]确保开发环境和生产环境依赖完全一致。特别注意httpx[http2]——我们实测开启HTTP/2后与天气API的连接复用率从32%提升到89%减少了TCP握手开销。而uvicorn[standard]会自动安装uvloop和httptools比纯Python实现快3倍以上。部署时我们用uvicorn main:app --host 0.0.0.0:8000 --workers 4 --loop uvloop --http httptools其中--workers 4对应4个CPU核心每个worker内跑一个uvloop形成真正的多进程单线程异步模型。4.2 核心模块拆解如何设计可测试、可监控、可降级的异步服务一个健壮的异步服务不能只写async def handler()。我们按生产级标准拆成四层4.2.1 数据访问层DAL用asyncpg.Pool管理连接拒绝每次请求新建连接# dal/weather.py import asyncpg from typing import Optional, Dict, Any class WeatherService: def __init__(self, pool: asyncpg.Pool): self.pool pool async def get_forecast(self, city: str) - Optional[Dict[str, Any]]: # 使用prepared statement防止SQL注入且预编译提升性能 stmt await self.pool.prepare( SELECT data FROM weather_cache WHERE city $1 AND updated_at NOW() - INTERVAL 1 hour ) row await stmt.fetchrow(city) if row: return row[data] # 缓存未命中调用外部API见下节 return await self._fetch_from_api(city) async def _fetch_from_api(self, city: str) - Dict[str, Any]: # 实际调用HTTP客户端此处省略 pass关键点asyncpg.Pool是连接池不是单个连接。它内部维护一个连接队列pool.acquire()会从池中取一个空闲连接用完自动归还。我们设min_size10, max_size100避免连接数爆炸。实测表明连接池大小设为2 * CPU核心数时吞吐量最优——太少会排队太多会触发数据库连接数限制。4.2.2 HTTP客户端层用httpx.AsyncClient的limits和timeout做熔断# clients/weather_api.py import httpx from typing import Dict, Any class WeatherApiClient: def __init__(self): # 关键配置限制并发连接数防打爆上游 limits httpx.Limits( max_connections100, # 总连接数上限 max_keepalive_connections20, # 长连接保活数 keepalive_expiry60.0, # 长连接存活时间 ) # 全局超时连接10s读写30s总耗时60s timeout httpx.Timeout( connect10.0, read30.0, write30.0, pool60.0, ) self.client httpx.AsyncClient( base_urlhttps://api.weather.com/v3/, limitslimits, timeouttimeout, http2True, ) async def get_forecast(self, city: str) - Dict[str, Any]: try: response await self.client.get( weather/forecast, params{city: city}, # 请求级超时覆盖全局 timeouthttpx.Timeout(20.0), ) response.raise_for_status() return response.json() except httpx.TimeoutException: # 降级返回缓存或默认值 return {error: timeout, fallback: cached_data} except httpx.HTTPStatusError as e: # 处理4xx/5xx raise WeatherAPIError(fAPI error {e.response.status_code})这里limits和timeout是生命线。没有limits1000并发请求会瞬间创建1000个TCP连接上游API直接503没有精细timeout一个慢请求会拖垮整个event loop。我们线上配置后API错误率从12.7%降到0.3%其中92%的失败是可控的超时降级而非雪崩。4.2.3 业务逻辑层用asyncio.gather和asyncio.wait控制并发粒度# services/weather_service.py import asyncio from typing import List, Dict, Any from dal.weather import WeatherService from clients.weather_api import WeatherApiClient class AggregatedWeatherService: def __init__(self, weather_dal: WeatherService, api_client: WeatherApiClient): self.weather_dal weather_dal self.api_client api_client async def get_multi_city_forecast(self, cities: List[str]) - List[Dict[str, Any]]: # 方案1全部并发适合城市数≤50 tasks [self._get_single_city(c) for c in cities] return await asyncio.gather(*tasks, return_exceptionsTrue) async def _get_single_city(self, city: str) - Dict[str, Any]: try: # 先查缓存 cached await self.weather_dal.get_forecast(city) if cached: return {city: city, data: cached, source: cache} # 缓存未命中调用API api_data await self.api_client.get_forecast(city) return {city: city, data: api_data, source: api} except Exception as e: return {city: city, error: str(e), source: error} async def get_multi_city_forecast_with_limit(self, cities: List[str], max_concurrent: int 10) - List[Dict[str, Any]]: # 方案2分批并发适合城市数50防压垮自身 results [] for i in range(0, len(cities), max_concurrent): batch cities[i:imax_concurrent] batch_tasks [self._get_single_city(c) for c in batch] batch_results await asyncio.gather(*batch_tasks, return_exceptionsTrue) results.extend(batch_results) return resultsasyncio.gather是并发利器但要注意它会同时启动所有task。如果cities列表有1000个城市gather会瞬间发起1000个HTTP请求即使你设了httpx.Limitsevent loop也要处理1000个协程调度。所以我们在高负载场景用分批batching策略用max_concurrent10控制节奏。return_exceptionsTrue很重要——它让gather在某个task抛异常时不中断整个列表而是把异常对象放进结果列表方便后续统一处理。4.2.4 API路由层用FastAPI的BackgroundTasks做异步日志和监控# main.py from fastapi import FastAPI, BackgroundTasks, HTTPException, Depends from services.weather_service import AggregatedWeatherService from typing import List, Dict, Any app FastAPI(titleWeather Aggregator API) # 依赖注入FastAPI自动管理生命周期 app.on_event(startup) async def startup_event(): # 初始化连接池和客户端 app.state.weather_dal WeatherService(await asyncpg.create_pool(...)) app.state.api_client WeatherApiClient() app.on_event(shutdown) async def shutdown_event(): await app.state.weather_dal.pool.close() await app.state.api_client.client.aclose() app.get(/forecast/multi) async def multi_city_forecast( cities: List[str], background_tasks: BackgroundTasks, service: AggregatedWeatherService Depends(get_weather_service), ) - Dict[str, Any]: # 记录请求日志异步不阻塞主流程 background_tasks.add_task(log_request, cities) try: # 执行核心业务 results await service.get_multi_city_forecast_with_limit(cities) return {status: success, data: results} except Exception as e: # 统一错误处理 raise HTTPException(status_code500, detailfInternal error: {str(e)}) async def log_request(cities: List[str]): # 异步写日志用aiofiles或直接发到日志服务 import aiofiles async with aiofiles.open(/var/log/weather_api.log, modea) as f: await f.write(f[{datetime.now()}] Requested {len(cities)} cities\n)BackgroundTasks是FastAPI提供的轻量级异步任务队列适合日志、指标上报、通知等“fire-and-forget”场景。它不保证执行成功失败就丢弃但绝不阻塞主请求流。我们用它把每条请求记录异步写入文件避免磁盘IO拖慢API响应。4.3 部署与监控用PrometheusGrafana盯死event loop的每一毫秒异步服务的监控和同步服务完全不同。我们重点关注三个指标Event Loop Lagevent loop处理一个任务的平均延迟。理想值10ms50ms说明loop被CPU密集型任务或阻塞IO拖住。Task Queue Length等待执行的协程数量。持续100说明并发压力过大或下游响应慢。Connection Pool Usage数据库/HTTP连接池的使用率。90%意味着需要扩容。我们用aiometer库暴露metrics# metrics.py from aiometer import run_all import asyncio async def monitor_event_loop_lag(): loop asyncio.get_running_loop() # 测量loop处理一个空任务的耗时 start loop.time() await asyncio.sleep(0) # 让出一次控制权 lag loop.time() - start return lag # 在FastAPI中间件中定期采集 app.middleware(http) async def add_metrics(request: Request, call_next): start_time asyncio.get_event_loop().time() response await call_next(request) process_time asyncio.get_event_loop().time() - start_time # 上报process_time, status_code等 return responseGrafana面板里我们设置告警规则当event_loop_lag_seconds{jobweather-api} 0.05持续2分钟立即触发PagerDuty。上线后第一次告警我们发现是某个cpu_heavy_calculation没走run_in_executor修复后lag回归正常。这就是异步监控的价值它不告诉你“服务挂了”而是告诉你“哪里卡住了”让你精准定位。5. 常见问题与排查技巧实录那些文档里不会写的血泪教训5.1 “RuntimeWarning: coroutine xxx was never awaited”不是警告是内存泄漏炸弹这个警告看似温和实则是严重bug。它意味着你创建了一个协程对象但从未await或asyncio.run()它。Python不会自动执行也不会垃圾回收——因为协程对象持有对所有局部变量的引用导致这些变量永远无法被释放。真实案例我们有个定时任务函数# ❌ 危险代码 async def cleanup_old_data(): await db.execute(DELETE FROM logs WHERE created_at NOW() - INTERVAL 7 days) # 在同步cron job里调用 def cron_job(): cleanup_old_data() # 忘了await只创建协程对象不执行结果每天生成12个未执行的协程对象每个持有一个数据库连接引用7天后内存暴涨1.2GBOOM Killer干掉了进程。修复方案只有两个在同步代码中必须用asyncio.run(cleanup_old_data())或者把cron job本身改成异步用asyncio.run()包装提示在CI/CD流水线中加入pylint检查规则W0104statement has no effect能捕获这种async def调用未await的情况。5.2 “Task was destroyed but it is pending!”未完成的Task正在偷偷吃掉你的连接当event loop关闭时如Uvicorn重启所有未完成的Task会被强制取消。如果这些Task正在执行await httpx.AsyncClient.get()而HTTP连接还没关闭就会出现这个警告。更糟的是这些“幽灵连接”会一直挂在操作系统层面直到TCP超时通常几分钟耗尽你的文件描述符fd。排查步骤lsof -i -n -p pid查看进程打开的socket过滤TIME_WAIT状态如果数量1000基本确认是未清理的Task在on_shutdown事件中显式等待所有Task完成app.on_event(shutdown) async def shutdown_event(): # 取消所有还在运行的Task tasks [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] [t.cancel() for t in tasks] # 等待它们完成最多5秒 await asyncio.gather(*tasks, return_exceptionsTrue) # 关闭资源 await app.state.weather_dal.pool.close() await app.state.api_client.client.aclose()5.3 “SSL handshake failed”异步HTTP客户端的证书信任链陷阱httpx默认使用系统证书但某些Linux发行版如Alpine的ca-certificates包不完整。aiohttp会静默失败而httpx直接抛httpx.ConnectError。解决方案不是禁用SSL验证绝对禁止而是显式指定证书路径import ssl import httpx # 从系统加载证书 context ssl.create_default_context() context.load_verify_locations(/etc/ssl/certs/ca-certificates.crt) client httpx.AsyncClient(verifycontext)或者在Dockerfile中确保安装完整证书FROM python:3.11-slim RUN apt-get update apt-get install -y ca-certificates rm -rf /var/lib/apt/lists/* COPY . /app WORKDIR /app5.4 性能倒退为什么加了asyncQPS反而下降了30%这是最打击信心的问题。我们遇到过三次原因各不相同场景根本原因解决方案实测效果场景1在async def里混用requests.get()requests是同步阻塞会卡住整个event loop全面替换为httpx.AsyncClient场景2数据库查询用await conn.fetchrow(SELECT * FROM ...)但表没建索引慢查询阻塞连接池所有协程排队等连接添加复合索引CREATE INDEX ON weather_cache(city, updated_at)场景3asyncio.gather并发数设为1000但httpx.Limits.max_connections20980个协程在event loop里排队等连接调度开销巨大降低gather并发数至min(1000, httpx.Limits.max_connections * 2)实操心得异步不是银弹。它放大了所有底层问题。在优化async之前先用asyncpg的EXPLAIN ANALYZE查SQL用httpx的log_levelDEBUG看HTTP连接复用率用asyncio的loop.slow_callback_duration0.01找出耗时10ms的回调——这才是高手的调试路径。5.5 迁移路线图如何把一个10万行的Django项目渐进式异步化没人能一夜之间把整个Django项目async化。我们的路线图分四步每步可独立上线Step 1异步视图Django 4.1把高IO、低计算的视图如调用外部API改为async def用sync_to_async()包装数据库操作。from asgiref.sync import sync_to_async sync_to_async def sync_db_query(user_id): return User.objects.get(iduser_id) async def async_view(request): user await sync_db_query(123) # 安全 return JsonResponse({user: