企业官网建设流程全解析

直接开工吧。Django 3 时代最让人兴奋的一件事，就是官方终于把异步这条路趟开了。但真到自己动手，你会发现文档里说的“支持异步”和项目里能跑起来的“异步”之间隔着不少坑。这篇文章我就拿一个具体场景——Django 3 集成 aiomysql 写异步视图、再用 channels 打通 WebSocket——把整个落地方案从头到尾掰开揉碎讲清楚。

先说清楚这套组合解决什么问题：如果你的项目里有实时推送需求（站内信、工单状态流转、在线统计这类），同时又想在接口层享受异步 IO 带来的高并发收益，那光用 DRF 写同步接口是不够的。Django 3.1+ 支持 async 视图，但 ORM 的操作还是同步的；channels 能把 WebSocket 接进来，可如果里面查数据库用的是同步 ORM，一个慢查询就能把事件循环堵死。所以正确的做法是：异步视图配异步数据库驱动，WebSocket consumer 里也走异步查询，三条链路全部异步化，才能真正扛住并发。

这套方案适合谁？已经熟悉 Django 基本开发、想往异步方向升级的团队，或者被 I/O 瓶颈卡过、想先在一个模块里试点异步改造的人。下面整个操作我都基于 Django 3.2 LTS 版本，Python 3.9+ 环境，这也是目前生产环境里最稳的组合。

1. 整体思路拆解：为什么这三样必须放一起

先理清一个容易混淆的概念：Django 3.0 开始支持 ASGI，但“支持 ASGI”不等于“支持异步视图”。Django 3.1 才正式加入异步视图和异步中间件支持，3.2 版本把这个能力稳定下来。所以版本别低于 3.1，生产环境直接选 3.2 LTS。

再说 aiomysql。很多人第一反应是“Django 不是有 ORM 吗，为什么还要单独接一个异步 MySQL 驱动？”问题就在这：Django ORM 的查询是同步阻塞的，在 async 视图里直接调用会报错——不是逻辑错误，是运行时错误（SynchronousOnlyOperation）。虽然可以用 sync_to_async 包装一下，让同步 ORM 跑在线程池里，但线程池的并发上限、上下文切换开销都摆在那，追求高并发时反而成了瓶颈。所以我的选择是：需要极致异步性能的接口，直接用 aiomysql 写原生 SQL，绕开 ORM；不需要那么高并发的接口，继续用 DRF + 同步 ORM，完全没问题。这俩不是替代关系，而是共存的。

channels 这套就更好理解了。它本质上是把 Django 放在了 ASGI 协议层之上，让 WebSocket、HTTP 请求可以走同一套 ASGI 应用。channels 自带的 channel layer 机制（channels_redis）负责在不同 consumer 进程之间通信，实现广播、群组推送这些核心功能。没有 channel layer，WebSocket 只能做点对点通信，服务端主动推送就只能自己拿 Redis 的 pub/sub 硬写，工作量和坑都会多不少。

把三样放一起的原因，可以对比一下这几种选型：

我实际项目里测过，同步方案在 200 并发时数据库连接池和请求线程池都开始吃紧，切换成这套组合后，稳定在 800 并发问题不大。当然前提是 SQL 本身写得不是特别烂，缓存该加也加。

2. 环境准备与项目骨架搭建

2.1 依赖版本与安装顺序

先列一份经过验证的依赖清单，照着装不会翻车：

Python 3.9+ Django==3.2.18 channels==3.0.5 channels-redis==4.1.0 aiomysql==0.2.0 PyMySQL==1.0.2 uvicorn[standard]==0.20.0 redis==4.5.4

需要注意的一点是 channels-redis 和 redis 库的版本匹配。channels-redis 4.x 用的是 redis-py 4.x 接口，如果装成 redis 5.x 某些地方会有兼容问题，4.1.0 + 4.5.4 这个组合是我长期在用的，稳定。aiomysql 0.2.0 是个比较新的版本，之前用的 0.1.x 在连接池回收上有些小毛病，能上新版就上新版。

安装就一句话：

pip install django==3.2.18 channels==3.0.5 channels-redis==4.1.0 aiomysql==0.2.0 PyMySQL==1.0.2 uvicorn[standard]==0.20.0 redis==4.5.4

装完之后先别急着写代码，确认一下 versions，尤其是 channels 的版本。channels 2.x 和 3.x 的 ASGI 配置方式差得挺多，3.x 用ProtocolTypeRouter，2.x 用channel_routing，网上不少老教程已经把这两个搞混了。

2.2 settings.py 的关键配置

Django 3.2 项目创建之后，settings.py里需要把 channels 加进 INSTALLED_APPS，同时配置 ASGI 入口和 channel layer：

# settings.py 核心配置 INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'channels', # 新增 'myapp', # 自己的应用 ] # 指定 ASGI 入口 ASGI_APPLICATION = 'myproject.asgi:application' # Channel Layer 用 Redis CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('127.0.0.1', 6379)], "symmetric_encryption_keys": [SECRET_KEY], # 线上记得换独立 key }, }, }

这里有个细节：channel layer 的symmetric_encryption_keys我建议别直接复用 SECRET_KEY，单独生成一个 key 存到环境变量里。因为 channel layer 的流量会经过 Redis，如果 Redis 被攻破，SECRET_KEY 泄露带来的连带风险太大，分开存能隔离伤害。

再看数据库配置。aiomysql 不走 Django ORM 的 DATABASES 配置，但 Django 自身的同步操作（比如 migrate、admin）还需要用 PyMySQL，所以两块都要配：

# settings.py 数据库双轨配置 # 同步部分：给 Django ORM 和 migrate 用 DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'mydb', 'USER': 'myuser', 'PASSWORD': 'mypassword', 'HOST': '127.0.0.1', 'PORT': '3306', 'CONN_MAX_AGE': 60, } } import pymysql pymysql.install_as_MySQLdb() # 让 Django 能正常连 MySQL

这个install_as_MySQLdb()是必须的，因为 Django 3.x 的 MySQL 后端默认找 MySQLdb 库，而它已经停止维护了。装 PyMySQL 后在__init__.py或者 settings 顶部调用一下，就能让 Django 正常使用 MySQL 数据库执行 migrate 等命令。

2.3 asgi.py 的正确写法

Django 3.2 创建项目时默认生成的是 wsgi.py 和 asgi.py，但默认的 asgi.py 只支持 HTTP。要用 channels，就得自己重写成 ProtocolTypeRouter 格式：

# myproject/asgi.py import os import django from django.core.asgi import get_asgi_application os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') django.setup() # 这行很重要，初始化 Django 应用 from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from channels.security.websocket import AllowedHostsOriginValidator import myapp.routing application = ProtocolTypeRouter({ "http": get_asgi_application(), "websocket": AllowedHostsOriginValidator( AuthMiddlewareStack( URLRouter( myapp.routing.websocket_urlpatterns ) ) ), })

这里面有两个很容易踩的坑。

第一个，django.setup()必须在导入任何 models 或 ORM 相关代码之前调用。如果你把 routing 模块的导入放在 django.setup() 前面，导入时它会尝试访问模型，报出类似 RuntimeError: Model class 不存在或 AppRegistryNotReady 的错误。

第二个，AllowedHostsOriginValidator不是可有可无的装饰。WebSocket 的 Origin 校验可以防止其他网站通过 JS 向你的 WebSocket 端点发起跨域连接。开发环境如果遇到 Origin 校验失败，记得在 ALLOWED_HOSTS 里加上对应域名，别图省事直接去掉这个校验器。

3. aiomysql 异步数据库访问层封装

3.1 为什么不用 Django ORM 的 sync_to_async

官方文档其实提供了一个异步访问 ORM 的便捷方法：sync_to_async(User.objects.get)。但我的实测数据告诉我们，这条路在高并发场景下不够用。核心原因有三点：

• sync_to_async 默认用线程池执行器，线程切换和 GIL 竞争还是有开销的；
• 线程池是共享资源，如果你的异步视图里有多个数据库查询，多个查询请求会争抢线程池名额，有可能互相排队；
• 当并发上来以后，线程池空闲数量不足，新的请求还得排队等线程，本质还是同步阻塞。

更重要的是，你既然已经选择了这一套纯异步链路（uvicorn 跑事件循环），任何一个环节出现同步阻塞，整个事件循环都被拖慢。就像一个流水线上只要有一个工位是手工操作，自动化再快也没用。

所以我在这套方案里，对需要高并发的接口直接用 aiomysql 访问数据库。aiomysql 是基于 asyncio 重写的 MySQL 驱动，完全没有同步代码，支持真正的异步连接池，这才符合“全链路异步”的初衷。

3.2 连接池封装：一个可以直接用的模块

先写一个连接池封装，放在myapp/db_pool.py：

# myapp/db_pool.py import aiomysql from django.conf import settings _pool = None async def get_pool(): global _pool if _pool is None: _pool = await aiomysql.create_pool( host=settings.DATABASES['default']['HOST'], port=settings.DATABASES['default']['PORT'], user=settings.DATABASES['default']['USER'], password=settings.DATABASES['default']['PASSWORD'], db=settings.DATABASES['default']['NAME'], autocommit=True, pool_recycle=3600, minsize=1, maxsize=20, charset='utf8mb4', cursorclass=aiomysql.DictCursor, ) return _pool async def close_pool(): global _pool if _pool: _pool.close() await _pool.wait_closed() _pool = None

几个参数说明：

autocommit=True是必须的。aiomysql 默认不会自动提交事务，如果在异步视图里执行了 INSERT 或 UPDATE 却没用 commit，数据写不进去，还不好排查。而 Django 事务钩子对这层不生效，所以直接在连接池层开掉最省心。

pool_recycle=3600很关键。MySQL 的 wait_timeout 默认 8 小时，但在某些云数据库实例上可能被调短。连接池里的连接如果长时间空闲，可能被服务端断开，但连接池还认为它活着，下一次请求就会拿到一个死连接。pool_recycle 定期回收重置，避免这个问题。

minsize=1这样第一个异步请求进来时不需要临时新建连接（其实还是会新建，但至少池子有保底），maxsize=20是保守值，如果你的单机并发很高，可以往上调，但要注意 MySQL 的最大连接数配置（max_connections）。

3.3 异步查询函数：原始 SQL 的正确姿势

连接池有了，还需要一层业务查询封装。这里拿一个“查询用户工单列表”举例：

# myapp/async_queries.py from myapp.db_pool import get_pool async def fetch_tickets_by_user(user_id, limit=20): pool = await get_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute( """ SELECT id, title, status, created_at FROM app_ticket WHERE user_id = %s AND is_deleted = 0 ORDER BY created_at DESC LIMIT %s """, (user_id, limit) ) result = await cur.fetchall() return result async def create_ticket(user_id, title, content): pool = await get_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute( """ INSERT INTO app_ticket (user_id, title, content, status, created_at) VALUES (%s, %s, %s, 'pending', NOW()) """, (user_id, title, content) ) return cur.lastrowid

这里有个细节值得展开：aiomysql 的%s占位符和 PyMySQL 一样，SQL 注入的安全防护要依赖这个参数化查询，千万别图省事用 f-string 拼接 SQL。直接拼接不仅不安全，遇到字符串里有引号的情况还会把 SQL 搞坏。

async with pool.acquire() as conn是连接池的上下文管理用法。它从池里拿一个连接，用完自动归还。要注意的是连表查询或者分页统计尽量放在一条 SQL 里完成，避免异步环境下多次往返 MySQL，这个开销在异步里虽然比同步小，但也不是零成本。

如果你对 ORM 的模型定义有硬性要求（比如已有大量模型代码），也可以在 aiomysql 查出结果后用模型构造对象，但我的经验是这类查询最好直接返回 dict，让视图层再包装。因为 DictCursor 返回的每行就是一个 dict，直接拿来序列化成 JSON 非常顺滑，反而经过 ORM 构造后会丢掉一些字段的默认类型。

4. 异步视图的实现方式与踩坑记录

4.1 一个完整的 async 视图

视图层这边，Django 3.1+ 允许直接写async def视图。下面这个例子同时演示了 aiomysql 查询和 JSON 返回：

# myapp/views.py import json from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from myapp.async_queries import fetch_tickets_by_user, create_ticket async def ticket_list(request, user_id): if request.method != 'GET': return JsonResponse({'code': 405, 'message': 'Method Not Allowed'}, status=405) try: tickets = await fetch_tickets_by_user(user_id, limit=50) except Exception as e: # 生产环境记得记日志 return JsonResponse({'code': 500, 'message': 'Database Error'}, status=500) return JsonResponse({'code': 200, 'data': tickets}) async def ticket_create(request): if request.method != 'POST': return JsonResponse({'code': 405, 'message': 'Method Not Allowed'}, status=405) try: body = json.loads(request.body) user_id = body['user_id'] title = body['title'] content = body['content'] except (KeyError, json.JSONDecodeError): return JsonResponse({'code': 400, 'message': 'Invalid Request'}, status=400) try: ticket_id = await create_ticket(user_id, title, content) except Exception: return JsonResponse({'code': 500, 'message': 'Database Error'}, status=500) return JsonResponse({'code': 200, 'data': {'ticket_id': ticket_id, 'status': 'pending'}})

一个很容易被忽略的点：Django 对async def视图的 CSRF 处理、session 中间件等，行为和同步视图不同。查 ASGI 文档会发现，SessionMiddleware 在异步上下文里会用 thread_sensitive 模式去拿 session，这个模式其实是同步的，性能会打折扣。所以线上环境我建议对纯 API 类异步视图，要么不用 session，要么直接关掉 SessionMiddleware，改用 JWT 或者自定义 token 头认证。别让 session 拖垮你异步链路的性能。

CSRF 方面，Django 3.2 对 async 视图的 CSRF 支持其实已经不错了，但很多人习惯在视图上加 @csrf_exempt。如果你只是想测试，临时加可以；生产环境最好用 django-cors-headers 或者自定义中间件来处理，全局 exempt 会带来安全问题。

4.2 视图路由配置与启动方式

异步视图的 url 配置和普通视图没区别，直接高兴就行：

# myapp/urls.py from django.urls import path from myapp.views import ticket_list, ticket_create urlpatterns = [ path('api/tickets/<int:user_id>/', ticket_list), path('api/tickets/', ticket_create), ]

但启动方式变了。开发环境下直接用 uvicorn 跑，别再runserver（runserver 在 channels 版本下其实也可以跑，但对异步视图的支持没有 uvicorn 完整）：

uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4

生产部署用 gunicorn + uvicorn worker 也行，Daphne 也可以，但我的实际经验里 uvicorn 的并发表现和对 asyncio 生态的兼容性都优于 Daphne，尤其是有多个 WebSocket 连接时。gunicorn 的话，配置--worker-class uvicorn.workers.UvicornWorker。

4.3 异步视图里的常见陷阱

第一个坑：在 async 视图里直接调用 ORM。报错信息是SynchronousOnlyOperation，意思是“你不能在线程外调用只有同步线程能访问的代码”。很多人第一次看到这个错就会想到 sync_to_async，但注意 sync_to_async 只是把代码丢到线程池里执行，并没有真正解决异步场景下的连接池问题。如果你必须用 ORM，那就用 sync_to_async 包装，但别忘了 DATABASES 配置里加CONN_MAX_AGE，否则每次请求都要新建连接。

第二个坑：await后面的代码不一定是异步安全的。比如你在await之后直接访问request.user就会踩雷，因为 Django 的 auth 中间件是同步的，它在 async 视图里访问 user 时可能返回的是 AnonymousUser。解决办法是在请求进来时就取好 user：

async def my_view(request): user = request.user # 这样直接访问有问题 # 改为 from asgiref.sync import sync_to_async user = await sync_to_async(lambda: request.user)()

第三个坑，也是我最开始遇到的：在 async 视图里写日志。Django 的 logging 模块本身是同步的，毫秒级的开销通常无所谓，但如果你在日志里做了格式化、写文件，在超高并发下也会成为瓶颈。我自己后来是把日志通过 queue 丢给独立线程去写，彻底把日志 I/O 和请求 I/O 分离。

5. channels 接入 WebSocket：实时推送完整链路

5.1 consumer 的高并发查询写法

WebSocket consumer 是 channels 的核心。下面这个 consumer 实现了：客户端连接后建立 channel；收到客户端消息后，异步查询数据库并向该 channel 发消息；同时支持把事件广播到整个 group。

# myapp/consumers.py import json from channels.generic.websocket import AsyncJsonWebsocketConsumer from myapp.async_queries import fetch_tickets_by_user class TicketConsumer(AsyncJsonWebsocketConsumer): async def connect(self): # 从 URL 中获取 user_id self.user_id = self.scope['url_route']['kwargs']['user_id'] self.group_name = f'user_{self.user_id}_tickets' # 加入组，这样服务端可以主动推送 await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() # 连接建立后，立刻推送一次当前工单列表 tickets = await fetch_tickets_by_user(self.user_id) await self.send_json({'type': 'ticket_list', 'data': tickets}) async def disconnect(self, code): await self.channel_layer.group_discard(self.group_name, self.channel_name) async def receive_json(self, content, **kwargs): action = content.get('action') if action == 'new_ticket': # 异步创建工单并广播 ticket_id = await create_ticket( self.user_id, content.get('title'), content.get('content') ) await self.channel_layer.group_send( self.group_name, { 'type': 'ticket_created', 'data': {'ticket_id': ticket_id, 'user_id': self.user_id} } ) else: await self.send_json({'type': 'error', 'message': 'Unknown action'}) async def ticket_created(self, event): # 这个方法是 group_send 中 type 对应的 handler await self.send_json({ 'type': 'ticket_created', 'data': event['data'] })

consumer 里值得学习的点：

self.scope['url_route']['kwargs']从路由里拿参数，这与 Django 视图里的 kwargs 类似。拿用户 ID 拼 group_name 是非常常见的模式，这样同一个用户的所有连接都在同一个 group 中，向 group 发消息等于向该用户的所有连接推送。

group_add和group_discard一定要成对写。连接建立时加入组，断开时移出组。如果漏掉 discard，旧连接断开后仍然能收到 Redis channel layer 广播的消息，导致重复推送。这个坑在本地测试时不一定暴露，线上用户多刷新几次就会有明显的重复消息。

receive_json里的action分发是个好习惯。如果 WebSocket 只有一个功能，你可以不设计 action 字段，但一段时间后需求总会变多（改状态、评论、刷列表都走同一个 socket），没有 action 字段会让后面的扩展很难受。

5.2 WebSocket 路由配置与 URL

consumer 写好了，需要在routing.py里配置路由：

# myapp/routing.py from django.urls import path from myapp.consumers import TicketConsumer websocket_urlpatterns = [ path('ws/tickets/<int:user_id>/', TicketConsumer.as_asgi()), ]

然后在项目的 asgi.py 里去引入这个 routing（上面已经写了）。注意as_asgi()是 channels 3.x 的必备操作，不能再像 channels 2.x 那样传一个 consumer 类。

前端连接代码就简单了：

const ws = new WebSocket('ws://localhost:8000/ws/tickets/123/'); ws.onopen = () => { console.log('连接成功'); ws.send(JSON.stringify({ action: 'new_ticket', title: '测试工单', content: '内容' })); }; ws.onmessage = (event) => { console.log(JSON.parse(event.data)); };

需要提醒一点：跨域 WebSocket 在前端会受 Origin 校验，如果前端和后端不在同一个域名下，需要配置ALLOWED_HOSTS，并且在前面说的AllowedHostsOriginValidator中放行对应来源。不放行的话，浏览器控制台只会报连接失败，不会告诉你具体原因，排查起来比较费劲。

5.3 服务端主动推送：从异步视图发消息到 WebSocket

这套组合最大的一个优势就是：不必局限于 WebSocket 连接内才能推送。你可以从异步视图里拿到 channel_layer，主动往某个 group 推送事件。

比如工单状态被后台修改了，可以在普通 HTTP 异步视图里也推送通知：

# myapp/push_utils.py from channels.layers import get_channel_layer from asgiref.sync import async_to_sync def notify_ticket_update(user_id, ticket_id, new_status): channel_layer = get_channel_layer() # 注意：这里是在同步视图里调用，所以用 async_to_sync # 如果在 async 视图里，直接 await 即可 async_to_sync(channel_layer.group_send)( f'user_{user_id}_tickets', { 'type': 'ticket_updated', 'data': {'ticket_id': ticket_id, 'status': new_status} } )

然后 consumer 里再加一个对应的事件处理：

async def ticket_updated(self, event): await self.send_json({ 'type': 'ticket_updated', 'data': event['data'] })

这种方式的适用场景很广：后台管理界面改了数据，WebSocket 端实时刷新；订单支付回调进来，用户页面立即出结果。如果不用 channel layer 的 group 机制，你要么得在每台服务器上再维护一份连接状态，要么只能用前端轮询，实时性和开发成本都不理想。

6. URL 路由、会话与认证体系的配套改造

6.1 WebSocket 上的身份认证

WebSocket 的认证方式和 HTTP 不太一样。channels 3.x 默认提供了AuthMiddlewareStack，它会从 session 里读取用户。但这个机制有几个问题：第一，前面提到 session 在异步链路中是同步的；第二，移动端/小程序往往没有 session cookie，它们更常使用 token 认证。

所以实践中我更推荐自己写一个 token 认证中间件，从 WebSocket 的连接参数（query string 或 header）中读取 token，并挂载到 scope 上：

# myapp/ws_auth.py from channels.db import database_sync_to_async from django.contrib.auth.models import AnonymousUser class TokenAuthMiddleware: def __init__(self, inner): self.inner = inner async def __call__(self, scope, receive, send): query_string = scope.get('query_string', b'').decode() token = None for param in query_string.split('&'): if param.startswith('token='): token = param.split('=', 1)[1] break if token: # 这里按你的 token 体系查询用户 scope['user'] = await self.get_user_from_token(token) else: scope['user'] = AnonymousUser() return await self.inner(scope, receive, send) @database_sync_to_async def get_user_from_token(self, token): try: return UserProfile.objects.get(api_token=token).user except UserProfile.DoesNotExist: return AnonymousUser()

注意这里get_user_from_token是同步 ORM 查询，但因为只在连接建立时执行一次，用database_sync_to_async完全没问题，不会成为链路瓶颈。如果这个查询也希望异步化，可以改成用 aiomysql 查一张 user_token 映射表。

前端连接时带上 token 参数即可：

const ws = new WebSocket(`ws://localhost:8000/ws/tickets/${userId}/?token=${token}`);

URL 路由里直接传 user_id，再配合 token 校验，就能防止用户越权访问别人的工单流。这是第一层安全，实际项目中通常还要在 consumer 里再校验一次self.scope['user']和 URL 里的 user_id 是否一致。别嫌重复，两层校验是必要的——URL 是客户端传的，永远不可信。

6.2 多进程部署下 channel layer 的正确姿势

如果 uvicorn 起了多个 worker，WebSocket 连接会分散在不同 worker 里，channel layer 用 Redis 就是它们通信的关键。你需要在 settings 中确认 CHANNEL_LAYERS 的 Redis 配置能被所有 worker 访问到，host 别写 127.0.0.1，除非 Redis 也在同一台机器上。生产环境建议给 Redis 加密码，并单独跑一个实例，别和缓存共用，避免消息堆积拖垮业务。

Redis 作为 channel layer 的稳定性其实很让人放心，但如果 Redis 挂了，所有 WebSocket 的 group 操作都会失败。线上建议给 Redis 加哨兵或者集群模式，channels-redis 4.x 支持 Redis Sentinel 配置，具体配置可以参考官方文档。小项目的话，单点 Redis + 定时持久化也够用了，重连机制 channels-redis 已经帮你处理了。

7. 实测数据与实际性能对比

7.1 压测工具与方案

我习惯用hey或者locust做 HTTP 压测，WebSocket 部分用自写的 Python asyncio 脚本模拟并发连接。先看下不同方案下的实测对比：

hey -n 5000 -c 500 -m GET http://localhost:8000/api/tickets/1/

这个命令表示 5000 个请求，500 并发。同类接口分别测同步视图、sync_to_async 包装的视图、aiomysql 异步视图三种实现，数据如下：

数据说明两个问题。第一，sync_to_async 有改善，但线程池共享还是带来额外开销；第二，aiomysql 不仅快，CPU 占用反而更低，因为大部分时间在等 I/O，没有线程上下文切换的损耗。

WebSocket 这部分的压测我用的是一款模拟长连接的脚本，500 个客户端同时连接、同时收发消息，Redis channel layer 单个节点的情况下完全没丢消息，延迟稳定在 10ms 以内。当并发推到 2000 时才开始出现消息排队延迟，Redis 的 CPU 成为瓶颈。

7.2 性能瓶颈的定位思路

异步化之后如果还是慢，先别怀疑框架，用 cProfile 或者简单的日志打点定位。我遇到过的情况里，90% 的“异步变慢”问题出在这三个地方：

• MySQL 慢查询没有根治，异步只是让慢查询不阻塞其他请求，但该慢的还是慢；
• Redis 和 MySQL 在同一台机器上，网络 IO 互相干扰；
• 一次请求里多次串行查询数据库，每条 SQL 都在等上一个完成。

异步的指导思想是 I/O 密集操作并发执行，而不是把一个请求里的多次 I/O 串行化。如果你发现一个异步视图里有两三条 SQL，但它们之间没有依赖关系，可以用asyncio.gather并发执行：

import asyncio async def get_multiple_data(user_id): results = await asyncio.gather( fetch_tickets_by_user(user_id, limit=20), fetch_unread_count(user_id), fetch_notices(user_id) ) return { 'tickets': results[0], 'unread': results[1], 'notices': results[2], }

这样总耗时从三次 I/O 的耗时相加变成了三次 I/O 的最大耗时。这个优化有时候比折腾数据库索引还管用。

8. 常见问题与排查集锦

8.1 aiomysql 连不上数据库

这个坑最常见，表现形式是异步视图里ConnectionRefusedError或者TimeoutError，但同步 ORM 却没问题。

• 检查数据库的 bind-address 是否监听 0.0.0.0，有些 MySQL 默认只监听 127.0.0.1，外部 IP 连不上；
• 检查用户权限，MySQL 用户的 host 要包含应用服务器的 IP，'myuser'@'%'比'myuser'@'localhost'更常用；
• 检查密码和端口配置是否继承自 DATABASES 默认值，如果你在 settings 里改了端口，连接池这里也要同步改。

还有一个小众问题：某些云 RDS 要求 app 端必须走特定端口，比如 3306 被安全组拦截，应用连不上，但直接命令行能连。这种情况一般是安全组规则没放通内网 IP，查网络层。

8.2 WebSocket 连接一直 404 或 403

排除思路和 HTTP 路由差不多：

• 先确认 URL 匹配：ws/tickets/1/尾部没有斜杠也会导致 404，channels 的 path 规则和 Django 的 path 规则相同；
• 再确认allowed_hosts配置，WebSocket 连接时 Origin 如果不在 ALLOWED_HOSTS 里，就会 403；
• 最后看看 asgi.py 里 ProtocolTypeRouter 有没有把 http 和 websocket 都配好。

我遇到过最奇怪的一个 case：本地runserver能连，上了 server 就不行。最后发现是 nginx 没配置升级 WebSocket 的 headers，关键配置是：

location /ws/ { proxy_pass http://backend_server; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_set_header Host $host; proxy_read_timeout 3600s; }

如果没有Upgrade和Connection "upgrade"这两个 header，nginx 会直接把 WebSocket 当成普通 HTTP 转发，客户端就永远建立不了连接。

8.3 channel layer 报错 ChannelFull 或消息丢失

ChannelFull是 Redis channel layer 的缓冲区满了。默认容量不大，如果 group 里成员很多，群发消息量又大，就容易触发。解决办法是调整容量：

CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('127.0.0.1', 6379)], "capacity": 1500, # 默认是 100 "expiry": 10, # 消息过期时间，默认 60 }, }, }

capacity调高后内存会有一定占用，自己衡量。expiry默认 60 秒，如果你的消息推送有较高实时性要求，可以调小一点，避免客户端消费慢时拿到过期消息。

消息丢失的情况更常见于 Redis 内存淘汰策略。如果 Redis 设了maxmemory-policy allkeys-lru之类的策略，channel layer 里未消费的消息可能被 LFU/LRU 淘汰掉。一定要确保 channel layer 可能用到的 key 不在淘汰范围。最简单的方式是在同一个 Redis 上不要混用缓存和 channel layer，或者把淘汰策略改成noeviction。

8.4 consumer 里查询数据库报 SynchronousOnlyOperation

这个问题在 channels 里也会出现。consumer 如果用了同步 ORM 查询，又没加database_sync_to_async包装，就会触发这个错。channels 官方其实提供了database_sync_to_async，用法：

from channels.db import database_sync_to_async @database_sync_to_async def get_user_by_id(user_id): return User.objects.get(id=user_id)

但注意这是同步查询，WebSocket 频繁收发时频繁调用会有性能损耗。对高频率的 consumer 操作，直接换成 aiomysql 查询反而更干净。

8.5 前端收到的数据不是 JSON 而是文本

这个坑特别好排查，也很好避坑。channels 的AsyncJsonWebsocketConsumer自带 JSON 编解码，send_json和receive_json会自动处理。如果你用的是WebsocketConsumer（同步），send和receive需要手动json.dumps和json.loads。

很多教程混着写，导致代码里await self.send_json(json.dumps(data))，发送了双重编码的字符串，前端收到后还要JSON.parse(JSON.parse(data))才能用，就很别扭。用哪个 consumer 就统一用哪套 API，别混。

9. 这套架构的扩展思路

写完这套组合并不代表结束，架构上还可以继续铺开。

第一个方向是引入消息队列（例如 RabbitMQ 或 Kafka）作为 channel layer 后面的事件总线。channels 的 group_send 只适合在单机或小规模集群内传播消息，当你有多个独立服务（比如定时任务服务、后台管理系统、外部回调服务）需要向 WebSocket 推送时，不能全部挤在 Redis 的 channel layer 上。常见的做法是：业务事件通过消息队列订阅和分发，消费者拿到事件后调用 channel_layer.group_send，把消息广播到 WebSocket group。这样 channel layer 只负责 WebSocket 连接管理，重活留给消息队列。

第二个方向是给 aiomysql 的查询层做一个简单的 repository 抽象。我一开始直接把 SQL 写在 consumer 里，后来接口多了发现很多重复代码，而且 SQL 和业务逻辑纠缠在一起很难维护。后来把查询按实体拆成模块，类似仓储模式的轻量实现，每个函数只接收明确参数、返回 dict，这样测试和复用都方便很多。

第三个方向是部署层面。用 uvicorn 多进程部署时，你需要保证每个进程能连到同一个 Redis 实例，同时数据库连接池各自独立没问题。Worker 数不要盲目调大，因为每个 worker 自己也维护了一个 aiomysql 连接池（20 个连接），4 个 worker + 20 个连接池 = 80 个数据库连接，别忘了给 MySQL 留出余量。

10. 收尾前的一些经验

整套方案我从 Django 2.2 时代就开始折腾，当时得靠 tornado + django-websocket 这类外部轮子，能用的方案都有很多限制。Django 3 把 asgi、async view、channels 的兼容性理顺之后，这套组合才算真正能投入生产。如果你正在做 Django 项目的异步化改造，强烈建议从一个无关紧要的接口开始试点，跑通之后再逐步推广，不要一上来就全量重构，风险太大。

最后分享一个我自己折腾过很久的细节：如果你的 WebSocket consumer 里要访问 request.user，即使你能拿到 user 对象，也别在每个连接、每条消息里反复查库。我这里就直接把 user_id 从 URL 里取出来用，user 对象放在 scope 里只在需要时才去查，省掉了很多次无谓的数据库往返。类似的优化思想贯穿整个异步改造过程，就是“减少一切不必要的 I/O”，数据库也好，Redis 也好，消息队列也好，每一条链路上的等待都可能成为并发性能的来源。

希望这篇文章能让你少踩几个坑。这套组合涉及的环节不少，但只要链路理清了，Django 项目的实时化、异步化能力会有一个非常明显的提升。