后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载Channels 是一个在 Django 原生能力之上构建的异步扩展层它将 Django 的应用能力从纯 HTTP 扩展到 WebSocket、聊天协议、IoT 协议等需要长连接的场景并统一建立在 Python 的 ASGI 规范之上。本文以官方文档门户 docs/index.rst 为骨架逐条展开其 4.x 系列的组件构成、核心概念Consumer、Routing、Channel Layer与完整学习路径并结合本仓库源码给出可验证的实现细节。读完本文你将理解 Channels 的整体架构掌握从编写 Consumer、配置路由到启用跨进程通信与 Django 认证/会话集成的完整实战方案。什么是 Django Channels按照文档门户 docs/index.rst 的定位Channels 是一个“把 Django 的能力扩展到 HTTP 之外”的项目除了传统的 HTTP 请求它还能处理 WebSockets、聊天协议、IoT 协议等更多传输类型。它建立在 Python 的ASGI异步服务器网关接口规范之上并直接依托 Django 自带的原生 ASGI 支持传统 HTTP 仍然由 Django 负责处理Channels 让你可以选择用同步风格类似 Django 视图的写法或异步风格来处理其他类型的连接两种风格可以在同一个项目中共存由你按场景取舍。从源码看本仓库即 Channels 的 Django 集成层本体。channels/apps.py中定义了 Django 应用配置类ChannelsConfigname 为channelschannels/__init__.py中声明当前版本为4.2.0并导出默认 channel layer 别名DEFAULT_CHANNEL_LAYER default。文档中反复强调的一个设计理念是“turtles all the way down”一路向下都是乌龟Channels 对“应用”只有一种统一的抽象即使是最简单的Consumer相当于 Django 中的 view本身也是一个完整合法的 ASGI 应用可以单独运行。这意味着 HTTP、WebSocket、自定义协议在处理模型上完全一致URL 路由、中间件本质上也都是 ASGI 应用。版本系列说明当前为 4.xdocs/index.rst明确指出这份文档对应Channels 4.x 系列。如果需要查阅旧版本可以在文档左下角的版本选择器中切换到3.x、2.x或1.x。本仓库对应的实际版本为4.2.0见 channels/init.py仓库中 docs/releases 目录保留了从 1.0.0 到 4.2.0 的完整发行说明如4.0.0.rst、4.1.0.rst、4.2.0.rst等可用于追踪各版本的演进历史。4.x 系列延续了 Channels 2 时代确立的“单 ASGI 应用 路由 Channel Layer”架构同时深度融合了 Django 的原生异步视图与 ASGI 支持。项目组成四个组件各司其职docs/index.rst的 “Projects” 一节列出了 Channels 体系的四个组成部分它们各自独立成库组件角色说明ChannelsDjango 集成层即本仓库提供 Consumer、Routing、Channel Layer 管理、认证/会话集成等DaphneHTTP 与 WebSocket 终止服务器面向生产环境的 ASGI 协议服务器负责把网络连接翻译成 scope/事件流asgiref基础 ASGI 库提供async_to_sync、sync_to_async等同步/异步桥接工具是 Channels 的底层依赖channels_redisRedis Channel Layer 后端可选官方维护的生产级 channel layer 实现支持单机、分片与群组广播这些组件在源码中有清晰的印证例如 channels/consumer.py 导入asgiref.sync.async_to_sync来把异步的send桥接到同步消费者channels/db.py 基于asgiref.sync.SyncToAsync实现database_sync_to_async。文档门户覆盖的是整个系统的使用方式单个组件的发行说明与具体指令则在各自仓库中维护。文档学习路径Topics 与 Reference 两大板块文档门户通过两个 toctree 组织全部内容Topics主题指南introduction、installation、tutorial/index、topics/consumers、topics/routing、topics/databases、topics/channel_layers、topics/sessions、topics/authentication、topics/security、topics/testing、topics/worker、deploying、topics/troubleshootingReference参考asgi、channel_layer_spec、community、contributing、support、releases/index。下文按这份路线图的核心脉络展开并结合仓库源码逐项深入。入门基础Scopes 与 Events作用域与事件入门文档 把一次连接拆成两个层次scope作用域与一系列events事件。Scope描述单次连接的一组细节例如 HTTP 请求的路径、WebSocket 的源 IP、聊天机器人对应的用户。scope 在整个连接期间持续存在对 HTTP 它只存活一次请求对 WebSocket 它存活整个 socket 生命周期断开重连则变化对其他协议则由该协议的 ASGI 规范决定。Events在 scope 生命周期内不断发生的事件代表用户交互——发起 HTTP 请求、发送一个 WebSocket 帧等。应用每个 scope 只会被实例化一次然后持续接收该 scope 内的事件流并决定如何响应。HTTP 的典型流程是用户发起请求 → 打开一个http类型 scope含路径、方法、headers→ 发送http.request事件含请求体→ 应用处理并生成http.response事件返回浏览器并关闭连接 → 请求/响应完成scope 销毁。聊天机器人则相反一个 scope 可能对应整段对话期间反复产生chat.received_message事件应用可以回发零到多条chat.send_message事件。核心概念一Consumer——事件驱动的应用单元Consumer是 Channels 代码的基本单元它“消费事件”本质上是一个微型的 ASGI 应用。当请求或新连接到达时Channels 依据路由表找到对应的 consumer 并启动它的一个实例。与 Django view 不同consumer 是长生命周期的——它存活于整个 scope 的持续期间。Consumer 把“自己写事件循环”这件事抽象为“按事件类型写对应方法”每种协议的事件类型对应一个命名方法Channels 负责调度与并行执行。文档门户经 introduction给出的最小示例class ChatConsumer(WebsocketConsumer): def connect(self): self.username Anonymous self.accept() self.send(text_data[Welcome %s!] % self.username) def receive(self, *, text_data): if text_data.startswith(/name): self.username text_data[5:].strip() self.send(text_data[set your username to %s] % self.username) else: self.send(text_dataself.username : text_data) def disconnect(self, message): pass同步还是异步SyncConsumer 与 AsyncConsumerchannels/consumer.py 定义了两种基类AsyncConsumer_sync False所有事件处理方法必须是协程self.send也是协程SyncConsumer_sync True通过database_sync_to_async把dispatch放到线程池中执行send则用async_to_sync桥回主事件循环。核心调度逻辑在AsyncConsumer.__call__与dispatch中它把收到的消息交给get_handler_name见 consumer.py将消息type中的.替换为_得到方法名例如事件websocket.receive对应方法websocket_receive随后用await_many_dispatch同时监听“客户端事件流”和“channel layer 事件流”。如果消息没有type字段或 type 以下划线开头会抛出ValueError找不到对应 handler 则抛出ValueError(No handler for message type ...)。选择原则来自 topics/consumers.rst要调用 Django ORM 等同步阻塞代码时用SyncConsumer它会把整个 consumer 跑在线程里避免 ORM 查询阻塞整个服务器的事件循环只在确定会受益于异步例如用 HTTPX 并行抓取 20 个页面且只使用异步原生库时才用AsyncConsumer默认推荐SyncConsumer若确实要在 AsyncConsumer 中调用同步函数可使用asgiref.sync.sync_to_async调用 ORM 则应使用 channels/db.py 提供的database_sync_to_async适配器或使用 Django 异步 API如aget。as_asgi()类方法consumer.py返回一个 ASGI v3 可调用对象每次 scope 到来时实例化一个新的 consumer作用类似 Django 视图的as_view()并支持通过initkwargs传入实例化参数。生命周期与关闭StopConsumer当 socket 或连接关闭例如收到http.disconnect或websocket.disconnect应用实例需要做清理完成后应抛出 channels/exceptions.py 中的StopConsumer来干净地停止 ASGI 应用。若不抛出服务器会在应用关闭超时Daphne 默认 10 秒后强制终止并发出警告。若自行启动了后台协程务必在连接结束时同步关闭它们以免协程泄漏。通用 ConsumerWebSocket 与 HTTPchannels/generic/websocket.py 与 channels/generic/http.py 提供了类似 Django 通用视图的现成实现WebsocketConsumerchannels.generic.websocket.WebsocketConsumer同步connect/receive(text_data, bytes_data)/disconnect(close_code)self.accept()接受连接可带子协议self.accept(subprotocol)客户端提供的子协议列表在self.scope[subprotocols]self.close(code4123)可带自定义错误码关闭send支持文本帧text_data与二进制帧bytes_data。groups类属性会在连接/断开时自动把该通道加入/移出对应群组若配置了不支持群组的 channel layer 或未配置 layer非空groups会触发InvalidChannelLayerError。AsyncWebsocketConsumer同上但全异步async defawait。JsonWebsocketConsumer / AsyncJsonWebsocketConsumer自动对文本帧做 JSON 编解码receive_json(content)与send_json(content)可通过覆写encode_json/decode_json类方法定制异步版本中这两个方法也是 async。AsyncHttpConsumerchannels.generic.http.AsyncHttpConsumer实现handle(body)接收完整请求体用send_response(status, body, headers)返回响应需要更细粒度控制如长轮询、Server-Sent Events时可使用底层send_headers与send_body(body, more_body...)。Consumer 内的 Scope 访问consumer 把连接的 scope 存在self.scope其中包含类似 Djangorequest对象的信息scope[path]路径HTTP/WebSocket、scope[headers]原始请求头HTTP/WebSocket、scope[method]请求方法HTTP。开启认证中间件后可访问scope[user]URLRouter 会把 URL 捕获组放入scope[url_route]。scope 也是中间件放置附加属性的地方与 Django 中间件往request上添加内容的方式一致。核心概念二Routing——按 scope 分派应用channels/routing.py 中所有路由类本身都是合法的 ASGI 应用可以嵌套。关键限制路由只工作在scope层面而非单个event层面一条连接只能对应一个 consumer路由的职责是“决定这条连接交给谁”而不是把事件分散到多个 consumer。建议以ProtocolTypeRouter作为项目根应用传给协议服务器的那个其下再嵌套协议相关的路由。ProtocolTypeRouter按协议类型分派channels.routing.ProtocolTypeRouter接受一个字典协议类型名 → ASGI 应用依据scope[type]分发ProtocolTypeRouter({ http: some_app, websocket: some_other_app, })若scope[type]不在映射中会抛出ValueError(No application configured for scope type ...)见 routing.py。要把 HTTP 处理拆分成长轮询 handler 与 Django 视图可用 URLRouter 并把get_asgi_application()作为最后一条“匹配一切”的路由。URLRouter按 URL 路径分派channels.routing.URLRouter接受 Django URL 对象列表path()或re_path()用于路由http/websocket类型连接URLRouter([ re_path(r^longpoll/$, LongPollConsumer.as_asgi()), re_path(r^notifications/(?Pstream\w)/$, LongPollConsumer.as_asgi()), re_path(r, get_asgi_application()), ])URL 捕获组会写入scope[url_route]kwargs为命名组字典args为位置组列表命名组与位置组不能混用一旦匹配到命名组位置组即被丢弃。取用示例stream self.scope[url_route][kwargs][stream]从源码看routing.pyURLRouter 会正确处理root_path、去掉路径开头的/并通过path_remaining支持嵌套路由同时明确不支持 Django 的include()需要嵌套时应使用嵌套的 URLRouter 实例否则抛出ImproperlyConfigured。若内层路由被额外中间件包裹path()路由的嵌套可能失效见文档中的 Issue 说明。ChannelNameRouter按 worker 通道名分派channels.routing.ChannelNameRouter依据 scope 中的channel键分派channel类型 scope用于 Worker 模式ChannelNameRouter({ thumbnails-generate: some_app, thumbnails-delete: some_other_app, })如果 scope 缺少channel键它会抛出ValueError提醒只能用于 channel 类型消息见 routing.py。根应用asgi.py 与 ASGI_APPLICATION项目需要定义一个根应用并通过ASGI_APPLICATION设置类比 Django 的ROOT_URLCONF指向它路径解析实现在 routing.py 的get_default_application()。按 Django 惯例建议放在项目级asgi.py文件中。仓库在 docs/includes/asgi_example.rst 给出了完整可复制的示例import os from channels.auth import AuthMiddlewareStack from channels.routing import ProtocolTypeRouter, URLRouter from channels.security.websocket import AllowedHostsOriginValidator from django.core.asgi import get_asgi_application from django.urls import path os.environ.setdefault(DJANGO_SETTINGS_MODULE, mysite.settings) # Initialize Django ASGI application early to ensure the AppRegistry # is populated before importing code that may import ORM models. django_asgi_app get_asgi_application() from chat.consumers import AdminChatConsumer, PublicChatConsumer application ProtocolTypeRouter({ # Djangos ASGI application to handle traditional HTTP requests http: django_asgi_app, # WebSocket chat handler websocket: AllowedHostsOriginValidator( AuthMiddlewareStack( URLRouter([ path(chat/admin/, AdminChatConsumer.as_asgi()), path(chat/, PublicChatConsumer.as_asgi()), ]) ) ), })核心概念三Channel Layers——跨进程通信当系统变复杂不同的应用实例之间需要通信例如聊天室里一个实例收到消息要广播给房间内其他实例。除了轮询数据库Channels 引入了channel layer——对一组传输通道的低层抽象用于在不同进程间传递信息。每个应用实例拥有唯一的channel name并可以加入groups从而同时支持点对点与广播通信。Channel layers 是完全可选的不想要时把CHANNEL_LAYERS设置留空或置为{}即可。配置 CHANNEL_LAYERS通过 Django 的CHANNEL_LAYERS设置配置使用channels.layers.get_channel_layer()获取默认 layer若使用 consumer则自动以self.channel_layer形式提供。配置管理实现在 channels/layers.py 的ChannelLayerManager按BACKEND字符串导入后端类以CONFIG字典实例化并支持通过TEST_CONFIG提供测试用配置当CHANNEL_LAYERS设置变化时缓存会自动失效连接了 Django 的setting_changed信号。Redis Channel Layer官方推荐的生产用后端需安装channels_redis包CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }它支持单服务器与分片两种配置并支持群组。内存 Channel Layer随 Channels 内置适合测试与本地开发CHANNEL_LAYERS { default: { BACKEND: channels.layers.InMemoryChannelLayer } }文档明确警告内存 layer切勿用于生产——每个进程是独立的 layer无法跨进程通信多实例环境下会有性能问题并最终导致数据丢失。纯异步接口与同步桥接channel layer 的send()、group_send()、group_add()等方法默认都是异步函数必须await。从同步代码调用时需要asgiref.sync.async_to_sync包装from asgiref.sync import async_to_sync async_to_sync(channel_layer.send)(channel_name, {...})该发什么高层事件而不是低层网络操作channel layer 用于应用层到应用层的高层通信发送高层事件由对端 consumer 处理后再执行低层网络操作如把内容转成 WebSocket 帧发给客户端。典型的聊天广播模式await self.channel_layer.group_send( room.group_name, { type: chat.message, room_id: room_id, username: self.scope[user].username, message: message, } )对端 consumer 定义chat_message处理方法type 中的.转_chat.join→chat_join把事件转成 WebSocket 帧。注意事件类型命名建议加前缀如chat.以避免与协议事件冲突AsyncConsumer 树中的所有事件处理方法都必须是async defSyncConsumer 树则全部是同步def。单通道通信每个应用实例每个长连接 HTTP 请求或打开的 WebSocket对应一个 consumer 实例启用 channel layer 时consumer 会生成唯一的channel name并开始监听。self.channel_name可在connect时写入数据库、disconnect时删除实现“按用户寻址”。从 consumer 外部发送单通道消息from channels.layers import get_channel_layer channel_layer get_channel_layer() await channel_layer.send(channel_name, { type: chat.message, text: Hello there!, })Groups 广播系统Groups 是内建于部分 channel layer 的广播系统支持向命名群组添加/移除通道名并整体发送提供群组过期机制清理因 disconnect handler 未能执行如断电而残留的连接不支持枚举群组内通道——它是纯广播系统需要精确控制或获知在线用户时请自建或使用第三方方案。用法同步 WebSocket consumer 中需async_to_sync桥接from asgiref.sync import async_to_sync class ChatConsumer(WebsocketConsumer): def connect(self): async_to_sync(self.channel_layer.group_add)(chat, self.channel_name) def disconnect(self, close_code): async_to_sync(self.channel_layer.group_discard)(chat, self.channel_name) def receive(self, text_data): async_to_sync(self.channel_layer.group_send)( chat, { type: chat.message, text: text_data, }, ) def chat_message(self, event): self.send(text_dataevent[text])群组名在默认后端中限制为 ASCII 字母、数字、连字符与句点最长 100 字符。在 consumer 之外如管理命令中发送消息时用get_channel_layer()获取 layer 后同样遵循“纯异步接口”规则在异步上下文中直接await否则用async_to_sync包装。核心概念四Django 集成——会话与认证Channels 为 Django 的常用功能提供了即插即用支持。把认证中间件包在路由外层即可让 WebSocket 等连接获得用户信息from django.core.asgi import get_asgi_application from django.urls import re_path django_asgi_app get_asgi_application() from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from channels.security.websocket import AllowedHostsOriginValidator application ProtocolTypeRouter({ http: django_asgi_app, websocket: AllowedHostsOriginValidator( AuthMiddlewareStack( URLRouter([ re_path(r^front(end)/$, consumers.AsyncChatConsumer.as_asgi()), ]) ) ), })AuthMiddlewareStackchannels/auth.py默认由CookieMiddlewareSessionMiddlewareAuthMiddleware组成会把scope[user]填充为当前用户取不到时为AnonymousUser。源码中get_user()通过database_sync_to_async桥接 ORM 读取会话里的 user_id 与认证后端并进行会话认证哈希校验constant_time_compare且会要求 scope 中先存在session即外层需有SessionMiddleware。AllowedHostsOriginValidatorchannels/security/websocket.py校验 WebSocket 请求的 Origin 头是否在 Django 的ALLOWED_HOSTS中用于阻止跨站 WebSocket 劫持。数据库访问、测试、Worker 与部署数据库topics/databases.rst 讲解在异步代码中安全使用 ORM。database_sync_to_asyncchannels/db.py是SyncToAsync的变体进入/退出线程时都会执行close_old_connections()清理旧连接aclose_old_connections()则是异步版本的连接清理工具consumer 的dispatch每次分发前都会调用它。测试topics/testing.rst 提供channels.testing工具如WebsocketCommunicator、HttpCommunicator、ApplicationCommunicator见 channels/testing可脱离真实服务器驱动 ASGI 应用进行断言内存 channel layer 常用于测试环境。Workertopics/worker.rst 介绍把 channel layer 用作轻量任务队列通过 management/commands/runworker.py 提供的runworker管理命令启动独立进程配合ChannelNameRouter按通道名路由任务消息。部署deploying.rst 说明用 Daphne 作为生产服务器与本文档体系中的 docs/asgi.rst ASGI 说明对应并涵盖ASGI_APPLICATION等设置的配置要点。故障排查topics/troubleshooting.rst 汇集常见问题例如同步代码误阻塞事件循环、channel layer 未配置导致的群组异常等。参考文档ASGI 与 Channel Layer 规范Reference 板块提供了两份重要的规范说明docs/asgi.rst介绍 ASGI 规范本身scope/event 结构、receive/send调用约定帮助理解 Channels 与协议服务器的接口边界。与 WSGI 类似ASGI 允许你在不同服务器和框架间自由选择而不被锁定在 Channels 与 Daphne 上。docs/channel_layer_spec.rst定义 channel layer 后端的接口规范如send、receive、new_channel、group_add等方法语义是编写自定义 channel layer 或接入第三方实现的依据ChannelLayerManager正是按此规范实例化后端。从文档到仓库继续深入的关键入口文档门户与学习路线docs/index.rst、docs/introduction.rst、docs/installation.rst、docs/tutorial/index.rstConsumer 实现channels/consumer.py、channels/generic/websocket.py、channels/generic/http.py路由实现channels/routing.py完整示例见 docs/includes/asgi_example.rstChannel Layerchannels/layers.py配置与用法见 docs/topics/channel_layers.rstDjango 集成channels/auth.py、channels/sessions.py、channels/security/websocket.py、channels/db.py测试与 Workerchannels/testing、channels/management/commands/runworker.py发行说明docs/releases/index.rst综上Django Channels 4.x 的核心价值在于把 Django 从“HTTP 视图框架”扩展为“任意协议的异步应用框架”同时保持同步与异步双轨制、提供跨进程通信的 channel layer并深度复用 Django 的认证、会话、ORM 与中间件体系。按本文给出的路径逐层深入你就能从文档导航走向可运行的实战项目。赞分享后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载相关推荐Django Channels 4.x 安装指南从 pip 安装到 ASGI 路由接入 Django 项目Django Channels 4.x 安装指南从 pip 安装到 ASGI 路由接入 Django 项目 本文是 Django Channels 4.x 的后端WebSocket异步编程Django Channels 入门指南用 ASGI、消费者与频道层为 Django 项目开启 WebSocket 与多协议实时能力Django Channels 入门指南用 ASGI、消费者与频道层为 Django 项目开启 WebSocket 与多协议实时能力 Channels 是在后端WebSocket异步编程革命性Wi-Fi软件定义无线电mobisys2018_nexmon_software_defined_radio项目深度解析革命性Wi Fi软件定义无线电mobisys2018_nexmon_software_defined_radio项目深度解析 mobisys2018_nexm上一篇抖音批量下载终极指南如何高效保存无水印视频内容下一篇lnd v0.16.3 要点解析mempool 效率优化、Macaroon 密钥重加密、通道链接重连与 HTLC 一致清扫创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考