Django3写异步接口有个天然的矛盾它从Django 3.0开始原生支持ASGI但ORM层面却一直没把异步做完整。我在一个实时业务监控项目里就被这个问题卡了很久——后端Django2 轮询撑不住多客户端同时刷新数据库查询稍微一慢整个请求队列就堵住。后来我把Django升到3.x用aiomysql接异步视图处理高频查询再用channels把推送链路改成WebSocket这套组合跑了快一年才敢说把这里面的坑都摸清了。这篇文章就把整个落地过程、选型逻辑和踩坑记录一次性讲透适合已经写过Django、想在现有项目上加异步查询和实时推送的开发者参考。1. 整体设计为什么要做Django3 aiomysql channels这套组合1.1 最初遇到的真实瓶颈当时项目场景是运营后台的实时大屏页面上一堆图表需要每2秒刷一次数据。旧方案纯粹靠前端定时器调HTTP接口请求量一上来MySQL连接和Django请求线程一起吃紧。更尴尬的是很多查询本身不重但每个请求都要走一遍Django中间件、ORM映射、序列化开销反而比SQL本身大。我想要的其实是两件事第一让数据库IO密集的操作别阻塞Django的事件循环第二把面向大屏的数据推送从“短轮询”改成“长连接”。前者需要异步视图 异步数据库驱动后者需要WebSocket通道。Django 3刚好都给了入口——ASGI支持异步视图channels框架负责WebSocket层。1.2 技术方案的选择范围市面上能做的组合其实不止一种我梳理过几个方向同步视图 轮询改动最小但治标不治本。DRF 异步Celery推送适合后台异步任务不适合实时双向通信。Django3 channels channels_redisWebSocket链路没问题但查询还是同步的。Django3 aiomysql async视图 channelsHTTP和WebSocket都能异步数据库访问也异步链路完整。最终选了最后一套原因很实际项目的业务逻辑已经从ORM迁走了一部分某些高频查询直接写SQL更可控aiomysql能复用原生asyncio事件循环不自带复杂的ORM映射开销。WebSocket端用channels是因为它是目前Django生态里最成熟的一套路由注册、鉴权、群组推送都内置好不必自己从头造协议解析。1.3 这套架构最终解决的核心问题组合完成之后前后性能差距非常直观。原来2秒轮询一批接口到多客户端并发时查询RT经常飙到300ms以上改造后异步视图的平均查询耗时能稳定在30ms上下大屏数据通过WebSocket通道实时推送客户端的“刷新”动作几乎变成了被动接收。这套方案对下面这些场景特别合适实时业务监控、运维大屏、交易看板。需要高频查询但SQL本身不复杂的内部系统。已经重度使用Django、暂时不打算换框架的团队。2. 核心细节解析Django3异步视图与aiomysql的配合原理2.1 Django3的异步视图到底改变了什么Django 3.0开始引入ASGI支持意味着视图可以定义成async def请求处理可以放进事件循环里跑。但有一个关键点容易忽略Django本身的ORM在3.x版本里并没有完全异步化早期版本的ORM查询在异步视图里直接调用会触发同步IO从而把事件循环卡住。所以当时最合理的做法是绕开ORM直接在异步视图里使用aiomysql操作数据库。异步视图处理HTTP请求aiomysql负责非阻塞数据库查询两者都在同一个asyncio循环里跑IO等待时事件循环可以继续处理其他请求。理解这一点就理解了整套架构为什么能缓解并发瓶颈——它并没有减少数据库的负担而是消除了Django同步进程等待IO时产生的空闲占用。2.2 aiomysql在这里扮演的角色aiomysql是一个纯asyncio的MySQL驱动核心思路和主流异步库一样用协程封装MySQL协议网络操作的等待不再占线程。它提供两种使用姿势aiomysql.connect()每个查询独立建连适合低并发调试。aiomysql.create_pool()维护连接池适合生产使用。我在项目里只用了连接池模式。每次查询从池里借连接、执行SQL、归还连接连接复用率高握手开销被压到最低。连接池还能设置最大连接数避免数据库被打满。建议把连接池封装成模块级单例在ASGI应用启动时初始化不要每次视图执行都create_pool否则连接池就失去意义了。封装大概是这样的import aiomysql from django.conf import settings _pool None async def init_db_pool(): global _pool if _pool is None: _pool await aiomysql.create_pool( hostsettings.DATABASES[default][HOST], portsettings.DATABASES[default][PORT], usersettings.DATABASES[default][USER], passwordsettings.DATABASES[default][PASSWORD], dbsettings.DATABASES[default][NAME], minsize5, maxsize20, autocommitTrue, charsetutf8mb4, ) return _pool2.3 异步视图和aiomysql衔接的写法写异步视图时逻辑和同步视图几乎一样只是函数变成async def。下面是一段高频查询接口的示例直接查最近5分钟的监控计数用信号量限制并发防止瞬间大量请求同时打到数据库import asyncio from django.http import JsonResponse from .db import get_pool _sem asyncio.Semaphore(10) async def metric_recent(request): async with _sem: pool await get_pool() async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute( SELECT COUNT(*) AS cnt FROM metrics WHERE created_at NOW() - INTERVAL 5 MINUTE ) row await cur.fetchone() return JsonResponse({count: row[cnt]})这里有几个细节容易出错我说一下每个协程用完cursor和连接都要释放aiomysql虽然会自动回收但显式关闭更稳妥。连接池的autocommit最好设为True避免事务残留导致连接池里的连接状态不干净。并发控制用Semaphore不然高峰期几百个协程同时借连接连接池再多也容易排队超时。我实际测试过改造前同步视图在200并发下大概有15%请求超时改造后这组异步视图在300并发下还能稳定响应效果比较明显。3. Windows本地环境配置从安装到跑起来3.1 安装依赖的版本匹配问题我本机是Windows开发环境依赖安装阶段就遇到一个经典坑channels和aiomysql的依赖包在Windows下安装容易出问题主要是加密扩展或编译环境。直接上我用的是这套版本组合经过实际验证稳定可用依赖版本说明Django3.2.x3.x稳定版ASGI支持完善channels3.0.xWebSocket核心库channels-redis4.0.x通道层Redis后端aiomysql0.0.22异步数据库驱动PyMySQL1.0.2aiomysql底层依赖必须装uvicorn0.20.xASGI服务器本地直接用这个daphne4.0.x如果走channels官方推荐就装这个命令行安装pip install django3.2.* channels3.* channels-redis4.* aiomysql0.0.22 pymysql1.* uvicorn[standard]Windows下如果遇到加密库编译报错换成预编译wheel包一般能解决。我这边直接用pip install cryptography先装好再装channels-redis就不会触发编译步骤。3.2 settings.py里需要做哪些修改Django3 channels的配置比较固定两块地方要动。第一是INSTALLED_APPS里把channels和daphne加在最前面官方建议daphne放在列表开头因为它要接管runserverINSTALLED_APPS [ daphne, django.contrib.admin, django.contrib.auth, ..., # 其他自带app channels, your_app, ]第二是ASGI应用路由和通道层的配置写在同一份settings里ASGI_APPLICATION myproject.asgi.application CHANNEL_LAYERS { default: { # 开发阶段先用内存通道层方便调试 BACKEND: channels.layers.InMemoryChannelLayer } }内存通道层只支持单进程生产环境要换Redis。我在开发阶段用内存层把WebSocket部分调通之后再切到Redis层测试多进程场景这两阶段的坑完全不一样后面会有专门小节讲。3.3 ASGI入口文件与路由组装项目里需要手写一个asgi.py入口文件Django3默认生成的asgi.py只有HTTP ASGI应用我们要手工扩展成HTTP和WebSocket共用。import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from your_app.routing import websocket_urlpatterns os.environ.setdefault(DJANGO_SETTINGS_MODULE, myproject.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: AuthMiddlewareStack(URLRouter(websocket_urlpatterns)), })ProtocolTypeRouter按协议区分流量分发HTTP请求走Django原生ASGI应用WebSocket连接走channels的URLRouter。AuthMiddlewareStack会从session里解析用户信息consumer里通过self.scope[user]就能拿到当前用户对象。如果不需要用户体系直接用URLRouter(websocket_urlpatterns)即可。4. channels实现WebSocket从路由到消费者4.1 WebSocket路由和消费者的基本结构路由文件可以类比Django的urls.pyconsumer则类比视图函数。我建了一个routing.pyfrom django.urls import re_path from your_app.consumers import MonitorConsumer websocket_urlpatterns [ re_path(rws/monitor/$, MonitorConsumer.as_asgi()), ]通行websocket协议同HTTP一样有路径概念这里的ws/monitor/就是客户端要连接的地址。消费者类继承AsyncWebsocketConsumer重写connect、receive、disconnect这三个生命周期方法。异步consumer内部跑在asyncio里能和aiomysql查询、通道层推送共用同一个事件循环比同步consumer更适合当前架构。一个连接建立后通常第一件事是把它加入一个房间或分组这样后端可以向整组推送消息。consumer里的写法import json from channels.generic.websocket import AsyncWebsocketConsumer class MonitorConsumer(AsyncWebsocketConsumer): async def connect(self): self.group_name monitor_group await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard(self.group_name, self.channel_name) async def receive(self, text_data): data json.loads(text_data) # 客户端发来的消息可以根据消息内容做不同回应 await self.send(text_datajson.dumps({ type: ack, message: received: data.get(msg, ), }))4.2 为什么建议用Redis做通道层内存通道层单进程内可以正常工作但真实部署环境往往是多进程甚至多节点消息要跨进程传递就得靠外部通道层。生产环境配置Redis版通道层的方式CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], capacity: 1500, expiry: 60, }, } }这里的capacity和expiry是排查消息丢失时的重要参数。capacity是单组消息队列上限超出的消息直接丢弃如果业务不允许丢消息需要调大或加持久化expiry是消息在Redis里存多久消费者长时间不消费消息就过期了。实时大屏场景下我们设置capacity1500基本够用。4.3 后端异步视图如何主动推消息给WebSocketchannels的魅力在于HTTP接口和WebSocket消费者能通过通道层联动。异步视图里可以直接拿到channel_layer执行group_send就能把消息推给所有在线客户端。from channels.layers import get_channel_layer from asgiref.sync import async_to_sync async def push_metric(request): channel_layer get_channel_layer() await channel_layer.group_send( monitor_group, { type: monitor.message, data: {count: 42}, }, ) return JsonResponse({status: pushed})注意type字段很关键它会映射到consumer里的同名方法把monitor.message的.message去掉转成monitor_message方法名。consumer里要对应实现async def monitor_message(self, event): await self.send(text_datajson.dumps({ type: metric, data: event[data], }))整条链路就是定时任务或业务逻辑调用异步视图 → group_send到Redis通道层 → Redis把消息投递给组内所有consumer → consumer把数据通过WebSocket发给浏览器。5. 开发环境跑通runserver到底在跑什么5.1 使用runserver直接启动settings里安装好daphne并配置好ASGI_APPLICATION后直接python manage.py runserver启动Django会自动使用daphne作为ASGI服务器。CtrlC停止无特殊参数。开发阶段如果发现WebSocket连接不上第一个检查项就是看终端有没有输出daphne相关的启动日志如果没有大概率是daphne没放在INSTALLED_APPS最前面或者根本没安装。前端连接WebSocket的js代码大概长这样const ws new WebSocket(ws:// window.location.host /ws/monitor/); ws.onmessage function(event) { const data JSON.parse(event.data); console.log(data); };5.2 runserver模式下同时调试HTTP和WebSocketdaphne接管runserver之后HTTP请求和WebSocket请求都走在同一个ASGI应用里。这意味着开发阶段完全不需要分别启动两个服务一个runserver就能同时调试API和推送链路和传统Django开发体验几乎没有差别这是channels方案比我之前用的独立socket服务方案顺手的地方。不过runserver默认是单进程的通道层如果用内存层一切正常如果切到Redis通道层单进程开发也能跑但要注意Redis服务必须提前启动否则connect之后receive会直接报通道层连接错误。5.3 开发环境的调试技巧开发时为了观察通道层消息流转我习惯在consumer里加一点简单的日志能看到连接建立、断开和消息推送时间点import logging logger logging.getLogger(__name__) async def connect(self): await super().connect() # 若重写了父类方法则手动调用 logger.info(websocket connected: %s, self.channel_name)另外一个开发时会踩的坑DEBUGTrue时Django会捕获异常并渲染调试页这在HTTP请求里很有用但WebSocket异常不会走HTML错误页而是直接断连。所以WebSocket接口出问题时优先看后端终端日志浏览器控制台只能看到“连接已断开”之类模糊信息。6. 生产环境部署从uvicorn到多进程通道层6.1 用uvicorn还是daphne生产环境我用的是uvicorn主要因为它的性能表现更稳定而且支持热重启。启动命令uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4如果你想保持channels原教旨也可以直接用daphnedaphne -b 0.0.0.0 -p 8000 myproject.asgi:application两种服务器都实现了ASGI规范都能处理WebSocket。差异主要在并发模型和易用性上uvicorn对多worker的支持更自然supervisor或systemd管理起来也顺手。6.2 多worker进程下的一个关键前提一旦--workers大于1内存通道层就必须换成Redis通道层。原因是每个worker是独立进程内存Queue没法和别的进程共享WebSocket消息只会发到所属进程的消费者那里另一个进程里的客户端永远收不到。我刚开始踩过这个坑单进程开发好好的一上多worker一半客户端收不到推送排查了半天才发现是通道层没有切换。生产环境通常放在反向代理后面比如NGINX。WebSocket要额外配置Upgrade和Connection头这一点在nginx里很容易漏。业务后端暴露的是HTTP服务和WS服务同一个ASGI端口nginx里location配置需要显式转发升级请求。6.3 数据库连接池规模的规划aiomysql连接池的大小要和uvicorn进程数配合。假设uvicorn起4个worker每个worker进程内部有一个独立的连接池如果池子maxsize设20那么数据库侧看到的连接数最多就是80。这个数字需要根据MySQL的max_connections来规划预留一部分给其他服务。我一般按这个公式估算总连接数 worker数量 × 池maxsize。如果数据库max_connections是200其他服务占用大概80那留给本服务的余量就是120池子maxsize定20到25比较合理。连接池不要开太大MySQL的连接开销不小。7. 常见问题与排查技巧实录7.1 问题速查表问题可能原因解决方案WebSocket连接后立即断开consumer的accept()没调用路由路径不匹配检查consumer里是否调用了await self.accept()消息推送丢失一半通道层仍为内存层但启动了多worker改用RedisChannelLayeraiomysql查询时报Event loop closed视图函数没有声明async def循环被意外关闭确认视图是async def连接池只在ASGI进程内初始化WebSocket连不上但HTTP正常ASGI_APPLICATION没配置反向代理未转发Upgrade头检查settings和nginx配置Redis连接数打满每个worker单独建Redis连接适当调大Redis的maxclients或减少worker数异步视图里调ORM卡住Django3.x ORM本身是同步的用aiomysql直连或sync_to_async包裹ORM调用7.2 我实际踩过的三个坑第一个坑是aiomysql连接池在runserver重载时泄漏。开发时修改代码触发runserver自动重载每个旧进程的连接池没有被立刻回收终端的错误信息会提示连接数过多。解决方式是尽量避免频繁重载或者自定义连接池状态检查定期关闭废弃连接。第二个坑是consumer里发消息前忘记检查连接是否已经关闭。客户端突然断网时self.send()会抛出异常如果不捕获协程直接退出。我后来统一在外面套一层异常捕获async def send_json_safe(self, data): try: await self.send(text_datajson.dumps(data)) except Exception: logger.warning(websocket send failed, connection may be closed)第三个坑是异步视图里直接调用Django ORM查询导致事件循环阻塞。很多人会把“Django3不是支持异步吗”理解成“ORM可以随便在异步视图里跑”其实Django 3.2的ORM操作大多还是同步实现在async def里执行会阻塞事件循环。要么用aiomysql这种全异步驱动要么用asgiref.sync.sync_to_async把ORM调用扔到线程池执行。from asgiref.sync import sync_to_async def get_orm_data(): # 这里是同步ORM查询 return list(MyModel.objects.filter(status1)[:20]) # 异步视图中这样调用 data await sync_to_async(get_orm_data)()7.3 关于性能调优的几点补充如果发现异步视图在高压下依然变慢先看是不是Semaphore设置太小。并发上限设太死请求会在信号量那里排队相当于人为制造瓶颈。建议把信号量放开到和连接池maxsize一致让所有可用连接都被充分复用。另一个性能隐患是每次查询都做字典转换。aiomysql的DictCursor虽然方便但每次fetch都会多一步字典构建高并发下浪费不少CPU。查询结果本身就三五字段的话用默认tuple cursor性能更好取值时按索引拿比如row[0]。最后补充一个监控提示建议给通道层加一个简单的监控脚本看group_send调用成功率。通道层消息丢失是实时推送系统最隐蔽的问题——Redis通道层在消息队列满时直接丢弃不会抛异常。我当时靠统计consumer收到的消息数量和业务侧发出的消息数量对比才发现capacity需要调大。这个对比逻辑现在已经成为项目的常规巡检项。这套Django3 aiomysql channels的组合从开发到生产跑下来最大的感受是Django做实时推送并不需要推倒重来只要你搞清楚异步视图用什么访问数据库、WebSocket消息怎么跨进程路由剩下的都是工程细节。我建议你在小项目上先跑通整条链路再逐步迁移业务毕竟数据一致性、消息可靠性和连接管理都只有在真实流量下才能真正检验出问题。