大家好,我是木木。
今天给大家分享一个实用的 Python 库,channels。
channels 是 Django 官方的异步扩展库,为 Django 项目带来 WebSocket,长轮询 HTTP 和后台任务支持,让 Django 从纯同步世界平滑迈入异步时代。
项目地址:https://github.com/django/channels
官方文档:https://channels.readthedocs.io
三大特点
官方维护
channels 是 Django 官方项目(Django Software Foundation),与 Django 核心版本保持同步兼容,提供稳定可靠的长期支持,不存在第三方维护的不确定性。
全协议支持
一套 channels 代码同时支持 HTTP、WebSocket 和 HTTP 长轮询三种协议。传统 Django view 写法继续有效,WebSocket 业务逻辑则由 Consumer 类处理,两套体系在同一应用中无缝共存。
灵活通道层
Channel Layer 支持 Redis、InMemory、Memory 等多种后端适配器,轻松实现跨进程消息推送和分布式 WebSocket 广播,配合 Daphne、Uvicorn 等 ASGI 服务器即可投入生产。
最佳实践
安装
pipinstallchannelschannels_redis
功能一:快速搭建 WebSocket Echo 服务
这段代码解决什么问题:使用 channels 快速实现一个 WebSocket 回显服务器,客户端发送什么内容,服务器原样返回。
# consumers.pyfromchannels.generic.websocketimportWebsocketConsumerimportjsonclassEchoConsumer(WebsocketConsumer):defconnect(self):self.accept()defreceive(self,text_data=None,bytes_data=None):# 收到什么就返回什么self.send(text_data=json.dumps({"type":"echo","data":text_data}))defdisconnect(self,close_code):pass
功能二:Routing 与 ASGI 配置
这段代码解决什么问题:配置 ASGI 入口文件,将 HTTP 请求和 WebSocket 请求分别路由到 Django view 和 Consumer,并加入 Auth 中间件支持 Django session。
# asgi.pyimportosfromchannels.routingimportProtocolTypeRouter,URLRouterfromchannels.authimportAuthMiddlewareStackfromdjango.core.asgiimportget_asgi_applicationos.environ.setdefault("DJANGO_SETTINGS_MODULE","myproject.settings")application=ProtocolTypeRouter({"http":get_asgi_application(),"websocket":AuthMiddlewareStack(URLRouter([# url(r"^ws/chat/$", EchoConsumer.as_asgi()),])),})
功能三:Channel Layer 广播实现群发
这段代码解决什么问题:使用 channels 的 Redis Channel Layer,在多个 WebSocket 连接之间实现消息广播,支持跨进程、跨服务器。
# consumers.pyfromchannels.generic.websocketimportAsyncWebsocketConsumerfromchannels.layersimportget_channel_layerfromasgiref.syncimportasync_to_syncclassChatConsumer(AsyncWebsocketConsumer):asyncdefconnect(self):self.room_name="global"self.group_name=f"chat_{self.room_name}"awaitself.channel_layer.group_add(self.group_name,self.channel_name)awaitself.accept()asyncdefreceive(self,text_data):awaitself.channel_layer.group_send(self.group_name,{"type":"chat_message","text":text_data})asyncdefchat_message(self,event):awaitself.send(text_data=event["text"])
Channel Layer 配置示例:
# settings.py — Channel Layer 配置(生产环境)CHANNEL_LAYERS={"BACKEND":"channels_redis.core.RedisChannelLayer","CONFIG":{"hosts":[("127.0.0.1",6379)],},}
高级功能:Consumer 方法体系
这段代码解决什么问题:channels 为 WebSocket 连接提供了完整生命周期管理,展示 AsyncWebsocketConsumer 的核心方法——connect/receive/disconnect。
classMyConsumer(AsyncWebsocketConsumer):asyncdefconnect(self):# 握手,可在此做鉴权awaitself.accept(subprotocol="graphql-ws")asyncdefreceive(self,text_data):awaitself.send(text_data="pong")asyncdefdisconnect(self,code):# code 为关闭码(1000 正常关闭)pass
channels 还支持 groups(组)、channel(命名通道)和定时任务,足以构建完整的实时应用后端。
环境与版本
- ASGI 服务器:Daphne / Uvicorn(需带 --websocket 标志)
适用
- Django 项目需要引入 WebSocket 实时通信
不适用
- 非 Django 项目(推荐直接使用 FastAPI + Starlette)
- 极致高性能纯 WebSocket 服务(推荐 uvicorn + 专用库)
- 轻量级即时通讯(推荐 socket.io 或 go-socket.io)
上线检查
- 确认使用 Daphne 或带 --websocket 的 Uvicorn 运行 ASGI 应用
- 生产环境务必配置 Redis Channel Layer,InMemory 仅适合开发
- WebSocket 鉴权在 connect() 方法中完成,拒绝未认证连接调用 self.close()
总结
channels 让 Django 开发者无需换框架,就能拥有完整的异步 WebSocket 能力。官方维护、协议全面、Channel Layer 灵活,是 Django 项目接入实时时代的最佳入口。