尧图网页设计 
后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载导读Channels 的 channel layers 不仅用于不同 ASGI 应用实例之间的实时通信还可以充当一套极简任务队列将耗时操作如生成缩略图、发送邮件、清理数据发送到固定名称的通道上由独立的 worker 进程监听并处理。本文以 docs/topics/worker.rst 为骨架结合仓库源码完整讲解事件的发送、消费者编写、路由配置与runworker命令的启动方式并阐明这套方案快而简单背后的设计取舍与使用边界。背景为什么用 channel layer 做任务队列在 docs/topics/channel_layers.rst 中channel layers 的定位是在不同 ASGI 应用实例之间通信的分布式消息通道。worker/后台任务系统正是对这一能力的复用把事件发往一组监听固定通道名的 worker 服务器从而构成一个低延迟的任务队列把重活从 Web 进程中卸载出去。值得强调的是Channels 的 worker 系统在设计上刻意保持简单且非常快代价是缺少一些你可能需要的特性没有重试机制消息最多被投递一次at-most-once delivery不保证任务一定被完成没有返回值生产者无法同步获取任务执行结果与内存型 channel layer 不兼容文档明确指出该功能在 in-memory channel layer 上不生效——因为内存层按进程隔离无法实现跨进程投递详见 channels/layers.py 中的InMemoryChannelLayer及其警告。因此官方建议对不需要完成保证的工作使用 Channels worker例如生成缩略图、清理缓存这类丢了也无伤大雅的任务如果需要更强的可靠性保证请改用独立的专用任务队列如 Celery。这也是后文所有配置示例中始终使用 Redis channel layer 的原因——worker 模式下必须使用跨进程的分布式后端例如官方维护的channels_redisCHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }第一部分发送事件Sending搭建后台任务分两步走发送事件与配置消费者接收并处理事件。先看发送侧。向固定通道名发送消息发送事件的方式非常直接——向一个固定的通道名调用send即可。以后台预生成缩略图为例在某个 Consumer 内部# Inside a consumer self.channel_layer.send( thumbnails-generate, { type: generate, id: 123456789, }, )这里有两个关键约定事件必须包含type键。即便该通道上只会发送一种类型的消息也必须带上type因为它最终会转变成消费者需要处理的一个事件。从 channels/consumer.py 的get_handler_name可以印证它首先检查消息是否含有type没有就抛出ValueError(Incoming message has no type attribute)然后把type中的点号替换为下划线作为方法名例如test.print→test_print。同理worker 侧在 channels/worker.py 的listener中也会校验message.get(type)缺失同样会报错。type的命名由你自己决定。发消息时你确定类型名写消费者时让方法名与之匹配.自动映射为_。从同步环境发送channel layer 的send是异步方法。如果发送动作发生在同步环境例如 Django 视图、management command 或同步 Consumer 中必须用asgiref.sync.async_to_sync包装这与 channel layers 文档中的用法一致from asgiref.sync import async_to_sync async_to_sync(self.channel_layer.send)( thumbnails-generate, { type: generate, id: 123456789, }, )在纯异步 Consumer 中则可直接await self.channel_layer.send(...)。第二部分接收与消费者Receiving and Consumersworker 事件在 scope 中的呈现方式Channels 会把到达 worker 的任务事件包装在一个type为channel、且带有与通道名一致的channel键的 scope 中呈现给应用。这一点可以从 channels/worker.py 的源码直接看到scope {type: channel, channel: channel}也就是说worker 模式下应用的 scope 类型是channel而不是http或websocket。用 ProtocolTypeRouter ChannelNameRouter 编排消费者官方推荐用ProtocolTypeRouter与ChannelNameRouter组合来编排你的消费者路由的完整说明见 docs/topics/routing.rstapplication ProtocolTypeRouter({ ... channel: ChannelNameRouter({ thumbnails-generate: consumers.GenerateConsumer.as_asgi(), thumbnails-delete: consumers.DeleteConsumer.as_asgi(), }), })从源码看这两者的分工ProtocolTypeRouter 根据scope[type]分发http、websocket、channel各路由到对应的子应用ChannelNameRouter 进一步根据scope[channel]分发到具体消费者若 scope 中没有channel键或通道名没有对应的应用都会抛出明确的ValueError。注意as_asgi()的作用它返回一个 ASGI 包装应用为每个 scope/连接实例化一个新的消费者对象相当于 Django 类视图的as_view()见 channels/consumer.py。编写事件处理消费者事件type由你在发送时自行命名因此消费者要与之匹配。比如一个基础消费者期望接收type为test.print的事件并从消息的text字段取出要打印的文本class PrintConsumer(SyncConsumer): def test_print(self, message): print(Test: message[text])其背后的调度机制位于 channels/consumer.pydispatch会根据get_handler_name(message)计算出的方法名test.print→test_print在消费者实例上查找对应方法并调用找不到处理器时会抛出ValueError(No handler for message type ...)。两个与消费者类型相关的硬性约束需要牢记同样来自 docs/topics/channel_layers.rst 的提示继承AsyncConsumer树时所有事件处理器包括 channel layer 事件必须是async def继承SyncConsumer树时处理器必须是同步defSyncConsumer会把dispatch放到线程池中执行见 channels/consumer.py。第三部分启动 workerrunworker消费者配置好之后还需要一个进程来喂它们。与 HTTP/WebSocket 不同worker 场景没有协议服务器、没有连接参与Channels 为此提供了runworker管理命令实现位于 channels/management/commands/runworker.pypython manage.py runworker thumbnails-generate thumbnails-deleterunworker 命令详解从 channels/management/commands/runworker.py 源码可以拆解出该命令的关键行为参数channels为必填的nargs位置参数即要监听的通道列表另有可选参数--layer指定使用哪个 channel layer 别名默认取DEFAULT_CHANNEL_LAYER即default定义于 channels/init.py。前置校验通过get_channel_layer()获取 layer若未配置任何CHANNEL_LAYERS直接抛出CommandError(You do not have any CHANNEL_LAYERS configured.)。应用来源通过get_default_application()从ASGI_APPLICATION设置中导入根应用见 channels/routing.py这要求你的项目正确配置了ASGI_APPLICATION指向包含ProtocolTypeRouter的asgi.py。执行实例化Worker并调用run()。最容易被忽略的坑runworker只会监听你在命令行上传入的通道。如果你漏掉了某个通道名或者干脆忘记运行 worker那么发往这些通道的事件就不会被接收、更不会被处理——任务会静静躺在 channel layer 里甚至按 layer 的expiry过期。因此发送方与接收方必须对通道名清单保持一致。Worker 底层实现每个通道一个监听协程channels/worker.py 的Worker类继承自asgiref.server.StatelessServer其handle()方法会为每个传入的通道名启动一个独立的监听协程async def handle(self): # For each channel, launch its own listening coroutine listeners [] for channel in self.channels: listeners.append(asyncio.ensure_future(self.listener(channel))) # Wait for them all to exit await asyncio.wait(listeners)每个listener是一个无限循环await channel_layer.receive(channel)取消息 → 校验type→ 构造{type: channel, channel: channel}的 scope → 通过get_or_create_application_instance取得或复用应用实例 → 把消息放入实例队列交给消费者处理。也就是说worker 进程是一个零连接、纯事件驱动的 ASGI 服务器消息到达即触发消费者对应的处理方法。第四部分生产实践要点与边界命名规范与容量控制通道名/组名的合法性由 channels/layers.py 中的BaseChannelLayer.valid_channel_name统一校验只允许 ASCII 字母、数字、连字符、下划线与句点长度小于MAX_NAME_LENGTH 100。设计任务通道名如thumbnails-generate、thumbnails-delete时请遵守该规则避免特殊字符。同时channel layer 支持通过CONFIG中的capacity与channel_capacity控制单个通道的容量上限见 channels/layers.py队列写满后继续send会抛ChannelFull异常channels.exceptions。如果后台任务量大可据此为任务通道配置专属容量防止积压冲垮内存或 Redis。用 ApplicationCommunicator 测试消费者worker 场景没有 HTTP/WebSocket 连接测试时适合用 channels/testing 提供的ApplicationCommunicator——它是面向任意 ASGI 应用的通用测试辅助文档见 docs/topics/testing.rst。可以构造type为channel的 scope直接向消费者send_input任务事件并断言其输出从而在不上线 Redis、不启动 runworker 的情况下验证处理器逻辑。方案取舍总结维度Channels worker 方案专用任务队列如 Celery延迟极低channel layer 直投取决于 broker 轮询策略可靠性at-most-once无重试支持重试、确认、结果存储返回值不支持支持适用场景缓存预热、缩略图、清理类可丢任务支付、邮件、数据同步等强保证任务channel layer 要求必须跨进程Redis 等内存层不可用依赖自身 broker总结Channels 的 worker/后台任务系统是 channel layers 能力的自然延伸生产者向固定通道名send带type的事件ChannelNameRouter把事件路由到对应消费者runworker进程则为每个通道启动监听协程并驱动消费者处理。整套链路在 channels/worker.py 与 channels/management/commands/runworker.py 中都有清晰的源码印证。它的定位是简单、极快、低延迟的轻量任务通道——使用时请牢记官方给出的边界最多一次投递、无重试无返回值、内存层不适用需要强保证的任务请交给专用任务队列。在此基础上配合 Redis channel layer 与规范命名即可为你的 Django/Channels 项目快速搭建一套实用的后台任务卸载机制。赞分享后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载相关推荐Redis延迟队列定时任务处理方案Redis延迟队列定时任务处理方案 在分布式系统中定时任务处理是常见需求如订单超时取消、定时通知等。传统定时任务方案存在精度不足、资源消耗大等问题。Red数据库缓存KV存储消息队列BepInEx.ConfigurationManager终极指南轻松管理游戏插件配置的免费工具BepInEx.ConfigurationManager终极指南轻松管理游戏插件配置的免费工具 你是否厌倦了手动编辑配置文件来调整游戏插件设置BepInEx游戏开发开发工具UI组件Activiti REST API开发指南构建流程引擎的远程服务接口Activiti REST API开发指南构建流程引擎的远程服务接口 Activiti REST API是Activiti流程引擎提供的强大远程服务接口它允后端示例工程上一篇AMD Ryzen硬件调试三大利器解锁专业级性能优化新境界下一篇5分钟快速上手Mermaid Live Editor免费在线图表编辑的终极解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Channels 后台任务与 Worker：用固定通道名实现低延迟任务队列
后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载导读Channels 的 channel layers 不仅用于不同 ASGI 应用实例之间的实时通信还可以充当一套极简任务队列将耗时操作如生成缩略图、发送邮件、清理数据发送到固定名称的通道上由独立的 worker 进程监听并处理。本文以 docs/topics/worker.rst 为骨架结合仓库源码完整讲解事件的发送、消费者编写、路由配置与runworker命令的启动方式并阐明这套方案快而简单背后的设计取舍与使用边界。背景为什么用 channel layer 做任务队列在 docs/topics/channel_layers.rst 中channel layers 的定位是在不同 ASGI 应用实例之间通信的分布式消息通道。worker/后台任务系统正是对这一能力的复用把事件发往一组监听固定通道名的 worker 服务器从而构成一个低延迟的任务队列把重活从 Web 进程中卸载出去。值得强调的是Channels 的 worker 系统在设计上刻意保持简单且非常快代价是缺少一些你可能需要的特性没有重试机制消息最多被投递一次at-most-once delivery不保证任务一定被完成没有返回值生产者无法同步获取任务执行结果与内存型 channel layer 不兼容文档明确指出该功能在 in-memory channel layer 上不生效——因为内存层按进程隔离无法实现跨进程投递详见 channels/layers.py 中的InMemoryChannelLayer及其警告。因此官方建议对不需要完成保证的工作使用 Channels worker例如生成缩略图、清理缓存这类丢了也无伤大雅的任务如果需要更强的可靠性保证请改用独立的专用任务队列如 Celery。这也是后文所有配置示例中始终使用 Redis channel layer 的原因——worker 模式下必须使用跨进程的分布式后端例如官方维护的channels_redisCHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }第一部分发送事件Sending搭建后台任务分两步走发送事件与配置消费者接收并处理事件。先看发送侧。向固定通道名发送消息发送事件的方式非常直接——向一个固定的通道名调用send即可。以后台预生成缩略图为例在某个 Consumer 内部# Inside a consumer self.channel_layer.send( thumbnails-generate, { type: generate, id: 123456789, }, )这里有两个关键约定事件必须包含type键。即便该通道上只会发送一种类型的消息也必须带上type因为它最终会转变成消费者需要处理的一个事件。从 channels/consumer.py 的get_handler_name可以印证它首先检查消息是否含有type没有就抛出ValueError(Incoming message has no type attribute)然后把type中的点号替换为下划线作为方法名例如test.print→test_print。同理worker 侧在 channels/worker.py 的listener中也会校验message.get(type)缺失同样会报错。type的命名由你自己决定。发消息时你确定类型名写消费者时让方法名与之匹配.自动映射为_。从同步环境发送channel layer 的send是异步方法。如果发送动作发生在同步环境例如 Django 视图、management command 或同步 Consumer 中必须用asgiref.sync.async_to_sync包装这与 channel layers 文档中的用法一致from asgiref.sync import async_to_sync async_to_sync(self.channel_layer.send)( thumbnails-generate, { type: generate, id: 123456789, }, )在纯异步 Consumer 中则可直接await self.channel_layer.send(...)。第二部分接收与消费者Receiving and Consumersworker 事件在 scope 中的呈现方式Channels 会把到达 worker 的任务事件包装在一个type为channel、且带有与通道名一致的channel键的 scope 中呈现给应用。这一点可以从 channels/worker.py 的源码直接看到scope {type: channel, channel: channel}也就是说worker 模式下应用的 scope 类型是channel而不是http或websocket。用 ProtocolTypeRouter ChannelNameRouter 编排消费者官方推荐用ProtocolTypeRouter与ChannelNameRouter组合来编排你的消费者路由的完整说明见 docs/topics/routing.rstapplication ProtocolTypeRouter({ ... channel: ChannelNameRouter({ thumbnails-generate: consumers.GenerateConsumer.as_asgi(), thumbnails-delete: consumers.DeleteConsumer.as_asgi(), }), })从源码看这两者的分工ProtocolTypeRouter 根据scope[type]分发http、websocket、channel各路由到对应的子应用ChannelNameRouter 进一步根据scope[channel]分发到具体消费者若 scope 中没有channel键或通道名没有对应的应用都会抛出明确的ValueError。注意as_asgi()的作用它返回一个 ASGI 包装应用为每个 scope/连接实例化一个新的消费者对象相当于 Django 类视图的as_view()见 channels/consumer.py。编写事件处理消费者事件type由你在发送时自行命名因此消费者要与之匹配。比如一个基础消费者期望接收type为test.print的事件并从消息的text字段取出要打印的文本class PrintConsumer(SyncConsumer): def test_print(self, message): print(Test: message[text])其背后的调度机制位于 channels/consumer.pydispatch会根据get_handler_name(message)计算出的方法名test.print→test_print在消费者实例上查找对应方法并调用找不到处理器时会抛出ValueError(No handler for message type ...)。两个与消费者类型相关的硬性约束需要牢记同样来自 docs/topics/channel_layers.rst 的提示继承AsyncConsumer树时所有事件处理器包括 channel layer 事件必须是async def继承SyncConsumer树时处理器必须是同步defSyncConsumer会把dispatch放到线程池中执行见 channels/consumer.py。第三部分启动 workerrunworker消费者配置好之后还需要一个进程来喂它们。与 HTTP/WebSocket 不同worker 场景没有协议服务器、没有连接参与Channels 为此提供了runworker管理命令实现位于 channels/management/commands/runworker.pypython manage.py runworker thumbnails-generate thumbnails-deleterunworker 命令详解从 channels/management/commands/runworker.py 源码可以拆解出该命令的关键行为参数channels为必填的nargs位置参数即要监听的通道列表另有可选参数--layer指定使用哪个 channel layer 别名默认取DEFAULT_CHANNEL_LAYER即default定义于 channels/init.py。前置校验通过get_channel_layer()获取 layer若未配置任何CHANNEL_LAYERS直接抛出CommandError(You do not have any CHANNEL_LAYERS configured.)。应用来源通过get_default_application()从ASGI_APPLICATION设置中导入根应用见 channels/routing.py这要求你的项目正确配置了ASGI_APPLICATION指向包含ProtocolTypeRouter的asgi.py。执行实例化Worker并调用run()。最容易被忽略的坑runworker只会监听你在命令行上传入的通道。如果你漏掉了某个通道名或者干脆忘记运行 worker那么发往这些通道的事件就不会被接收、更不会被处理——任务会静静躺在 channel layer 里甚至按 layer 的expiry过期。因此发送方与接收方必须对通道名清单保持一致。Worker 底层实现每个通道一个监听协程channels/worker.py 的Worker类继承自asgiref.server.StatelessServer其handle()方法会为每个传入的通道名启动一个独立的监听协程async def handle(self): # For each channel, launch its own listening coroutine listeners [] for channel in self.channels: listeners.append(asyncio.ensure_future(self.listener(channel))) # Wait for them all to exit await asyncio.wait(listeners)每个listener是一个无限循环await channel_layer.receive(channel)取消息 → 校验type→ 构造{type: channel, channel: channel}的 scope → 通过get_or_create_application_instance取得或复用应用实例 → 把消息放入实例队列交给消费者处理。也就是说worker 进程是一个零连接、纯事件驱动的 ASGI 服务器消息到达即触发消费者对应的处理方法。第四部分生产实践要点与边界命名规范与容量控制通道名/组名的合法性由 channels/layers.py 中的BaseChannelLayer.valid_channel_name统一校验只允许 ASCII 字母、数字、连字符、下划线与句点长度小于MAX_NAME_LENGTH 100。设计任务通道名如thumbnails-generate、thumbnails-delete时请遵守该规则避免特殊字符。同时channel layer 支持通过CONFIG中的capacity与channel_capacity控制单个通道的容量上限见 channels/layers.py队列写满后继续send会抛ChannelFull异常channels.exceptions。如果后台任务量大可据此为任务通道配置专属容量防止积压冲垮内存或 Redis。用 ApplicationCommunicator 测试消费者worker 场景没有 HTTP/WebSocket 连接测试时适合用 channels/testing 提供的ApplicationCommunicator——它是面向任意 ASGI 应用的通用测试辅助文档见 docs/topics/testing.rst。可以构造type为channel的 scope直接向消费者send_input任务事件并断言其输出从而在不上线 Redis、不启动 runworker 的情况下验证处理器逻辑。方案取舍总结维度Channels worker 方案专用任务队列如 Celery延迟极低channel layer 直投取决于 broker 轮询策略可靠性at-most-once无重试支持重试、确认、结果存储返回值不支持支持适用场景缓存预热、缩略图、清理类可丢任务支付、邮件、数据同步等强保证任务channel layer 要求必须跨进程Redis 等内存层不可用依赖自身 broker总结Channels 的 worker/后台任务系统是 channel layers 能力的自然延伸生产者向固定通道名send带type的事件ChannelNameRouter把事件路由到对应消费者runworker进程则为每个通道启动监听协程并驱动消费者处理。整套链路在 channels/worker.py 与 channels/management/commands/runworker.py 中都有清晰的源码印证。它的定位是简单、极快、低延迟的轻量任务通道——使用时请牢记官方给出的边界最多一次投递、无重试无返回值、内存层不适用需要强保证的任务请交给专用任务队列。在此基础上配合 Redis channel layer 与规范命名即可为你的 Django/Channels 项目快速搭建一套实用的后台任务卸载机制。赞分享后端WebSocket异步编程【免费下载链接】channelsDeveloper-friendly asynchrony for Django项目地址https://gitcode.com/gh_mirrors/ch/channels点击查看免费下载相关推荐Redis延迟队列定时任务处理方案Redis延迟队列定时任务处理方案 在分布式系统中定时任务处理是常见需求如订单超时取消、定时通知等。传统定时任务方案存在精度不足、资源消耗大等问题。Red数据库缓存KV存储消息队列BepInEx.ConfigurationManager终极指南轻松管理游戏插件配置的免费工具BepInEx.ConfigurationManager终极指南轻松管理游戏插件配置的免费工具 你是否厌倦了手动编辑配置文件来调整游戏插件设置BepInEx游戏开发开发工具UI组件Activiti REST API开发指南构建流程引擎的远程服务接口Activiti REST API开发指南构建流程引擎的远程服务接口 Activiti REST API是Activiti流程引擎提供的强大远程服务接口它允后端示例工程上一篇AMD Ryzen硬件调试三大利器解锁专业级性能优化新境界下一篇5分钟快速上手Mermaid Live Editor免费在线图表编辑的终极解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考