channels,一个实用的 Python 库

大家好，我是木木。

今天给大家分享一个实用的 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（命名通道）和定时任务，足以构建完整的实时应用后端。

环境与版本

• Python：≥ 3.9
• Django：≥ 4.2
• channels：4.x（当前最新稳定版）
• channels_redis：4.x
• ASGI 服务器：Daphne / Uvicorn（需带 --websocket 标志）

适用

• Django 项目需要引入 WebSocket 实时通信
• 需要长轮询 fallback 的企业内网场景
• 需要跨进程/跨服务器消息推送的分布式架构
• 已有 Django 项目平滑接入异步生态

不适用

• 非 Django 项目（推荐直接使用 FastAPI + Starlette）
• 极致高性能纯 WebSocket 服务（推荐 uvicorn + 专用库）
• 轻量级即时通讯（推荐 socket.io 或 go-socket.io）

上线检查

1. 确认使用 Daphne 或带 --websocket 的 Uvicorn 运行 ASGI 应用
2. 生产环境务必配置 Redis Channel Layer，InMemory 仅适合开发
3. WebSocket 鉴权在 connect() 方法中完成，拒绝未认证连接调用 self.close()

总结

channels 让 Django 开发者无需换框架，就能拥有完整的异步 WebSocket 能力。官方维护、协议全面、Channel Layer 灵活，是 Django 项目接入实时时代的最佳入口。