任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载导读celery call是 Celery 分发任务的核心 CLI 命令它允许你在不编写任何 Python 代码的情况下直接通过命令行向消息代理Broker发送任务非常适合调试、运维和脚本化调用场景。本文以 celery.bin.call 模块 为主线结合 celery/bin/call.py 的源码实现系统讲解该命令的参数语义、底层调用链、执行选项与路由选项的配置方法并给出可直接复制的实战示例。命令概览celery call是什么celery call是 Celery 命令行工具celery的子命令定义在 celery/bin/call.py 中其核心作用正如源码 docstring 所写Thecelery callprogram used to send tasks from the command-line.用于从命令行发送任务的程序。该命令通过click.command(clsCeleryCommand)注册并在 celery/bin/celery.py 中通过celery.add_command(call)挂载到主命令celery上因此你可以通过以下方式调用它celery call task_name [options]与celery的其他子命令如worker、beat、control不同call不需要启动任何本地进程它只做一件事将一条任务消息发送到 Broker并打印出生成的 task id。与delay()/apply_async()的关系从使用角度celery call等价于在 Python 中调用app.send_task()。二者的区别在于Task.delay()和Task.apply_async()是Task对象的方法要求任务已注册在当前进程中app.send_task()是Celery应用的方法通过任务名称发送不要求任务已注册——这正是celery call的底层实现所依赖的能力。Celery 官方文档 calling.rst 也明确指出如果任务没有在当前进程中注册可以使用send_task按名称调用任务。celery call把这个能力带到了 Shell 层。源码剖析命令是如何工作的celery call的实现非常精炼核心逻辑集中在 celery/bin/call.py 的call()函数中click.command(clsCeleryCommand) click.argument(name) click.pass_context handle_preload_options def call(ctx, name, args, kwargs, eta, countdown, expires, serializer, queue, exchange, routing_key): Call a task by name. task_id ctx.obj.app.send_task( name, argsargs, kwargskwargs, countdowncountdown, serializerserializer, queuequeue, exchangeexchange, routing_keyrouting_key, etaeta, expiresexpires ).id ctx.obj.echo(task_id)底层调用链整个命令的执行流程可以概括为参数解析Click 框架解析name位置参数和所有选项上下文获取click.pass_context将 Click 上下文传入ctx.obj是CLIContext实例定义于 celery/bin/base.py它持有app属性——即当前 Celery 应用发送任务调用ctx.obj.app.send_task(name, args..., kwargs..., ...)将命令参数透传给send_task输出结果send_task返回一个AsyncResult对象取其.id属性通过ctx.obj.echo(task_id)打印到终端。send_task内部发生了什么send_task是 Celery 应用的核心方法定义在 celery/app/base.py 的send_task方法约 L945 起。从源码看它的关键步骤包括生成 task id若未显式指定使用task_id task_id or uuid()生成 UUID按名称查找任务如果调用方未提供task_type会先在本地注册表self._tasks中查找同名任务若找到则合并其执行选项如time_limit、soft_time_limit等作为默认值——这里直接使用底层注册表避免send_task触发应用的自动 finalize创建任务消息将参数序列化后通过amqp.send_task_message(P, name, message, **options)将消息发布到 Broker返回结果对象构造result_cls or self.AsyncResult实例默认即AsyncResult把.id暴露给调用方。因此celery call打印出的 task id 就是这条消息在 Broker 中的唯一标识后续可以用celery result task_id查询执行结果或在 Python 中用AsyncResult(task_id)检索。注意源码中send_task对task_always_eager配置会发出AlwaysEagerIgnored警告因为send_task本身是纯消息发送不会同步执行任务。也就是说即使配置了task_always_eagerTruecelery call依然只发送消息不会在本进程内直接运行任务。位置参数name任务名称celery call唯一的位置参数是任务名称namecelery call proj.tasks.add任务名称必须是已注册或可被 Broker 路由识别的名称。对于本地注册的任务send_task会在注册表中找到它并合并其执行选项见上文源码分析对于未在本地注册的远程任务send_task同样允许按名称发送这是celery call用于跨应用调试的典型场景。Calling Options任务执行参数celery call提供了一组 Calling Options调用选项控制任务消息的载荷与调度行为全部定义在 celery/bin/call.py 中并通过help_groupCalling Options在帮助信息中分组展示。选项短参数类型默认值说明--args-aJSON 数组[]位置参数如[1, 2]--kwargs-kJSON 对象{}关键字参数如{x: 1}--eta—ISO 8601 时间无计划执行时间绝对时间--countdown—浮点数无从现在起多少秒后执行--expires—ISO 8601 时间或浮点数无过期时间过期后任务不再执行--serializer—字符串json任务消息序列化器--args与--kwargsJSON 参数解析--args和--kwargs分别使用JSON_ARRAY和JSON_OBJECT参数类型定义于 celery/bin/base.py 的JsonArray和JsonObject类JsonArray.convert()使用json.loads(value)解析并要求解析结果必须是 list否则报错f{value} was not an arrayJsonObject.convert()同理要求解析结果是 dict否则报错f{value} was not an object。因此参数必须以合法的 JSON 语法书写。例如# 位置参数 [2, 2]任务 add(2, 2) celery call proj.tasks.add -a [2, 2] # 关键字参数 {x: 1, y: 2} celery call proj.tasks.add -k {x: 1, y: 2}注意 JSON 字符串在 Shell 中通常需要用引号包裹避免花括号、方括号被 Shell 解释。--eta与--countdown定时执行--eta使用ISO8601参数类型ISO8601DateTime内部调用maybe_iso8601()定义于 celery/utils/time.py将 ISO 8601 字符串解析为datetime对象。例如# 在 2026-09-19 10:00:00UTC执行 celery call proj.tasks.add -a [2, 2] --eta 2026-09-19T10:00:00Z--countdown使用float类型表示从现在起多少秒后执行本质上是eta now countdown的便捷写法# 30 秒后执行 celery call proj.tasks.add -a [2, 2] --countdown 30--expires过期时间--expires使用ISO8601_OR_FLOAT参数类型ISO8601DateTimeOrFloat它先尝试将值解析为浮点数秒数失败后再尝试按 ISO 8601 时间解析——这种双模式设计让过期时间既可以用秒数表示也可以用绝对时间表示# 相对时间60 秒后过期 celery call proj.tasks.add -a [2, 2] --expires 60 # 绝对时间在指定时刻过期 celery call proj.tasks.add -a [2, 2] --expires 2026-09-19T12:00:00Z过期后的任务消息会被丢弃不会被执行。--serializer序列化器默认序列化器为json。Celery 支持多种序列化器如pickle、msgpack、yaml等具体可用的序列化器取决于安装的依赖。使用方式celery call proj.tasks.add -a [2, 2] --serializer jsonRouting Options消息路由参数celery call还提供一组 Routing Options路由选项用于控制任务消息发往哪个队列全部通过help_groupRouting Options分组选项说明--queue自定义队列名称--exchange自定义交换机名称--routing-key自定义路由键这三个选项直接透传给send_task的queue、exchange、routing_key参数。它们用于覆盖应用配置中的默认路由规则task_routes、task_default_queue等适合以下场景将任务发送到默认路由之外的特定队列配合自定义交换机/路由键将消息路由到特定的 Worker。# 发送到名为 high_priority 的队列 celery call proj.tasks.add -a [2, 2] --queue high_priority # 指定交换机与路由键 celery call proj.tasks.add -a [2, 2] --exchange my_exchange --routing-key my_key路由的底层实现在send_task中它使用router router or amqp.router即应用的消息路由对象最终通过amqp.send_task_message(P, name, message, **options)把路由选项应用到消息发布过程。全局选项与预加载选项celery call作为celery主命令的子命令还继承所有全局选项Global Options例如-A/--app指定 Celery 应用模块路径如-A proj--workdir切换工作目录-C/--no-color禁用彩色输出-q/--quiet静默模式--config指定配置模块--broker覆盖 Broker 连接 URL。此外call()函数被handle_preload_options装饰器包裹定义于 celery/bin/base.py。该装饰器会收集应用中通过user_options注册的preload选项并在执行命令前通过user_preload_options信号定义于 celery/signals.py发送给监听者。这意味着第三方插件可以通过信号机制在命令执行前注入自定义行为。完整实战示例假设你有一个 Celery 应用proj参考 examples/next-steps/proj/celery.py 的典型结构其中注册了任务proj.tasks.add# proj/tasks.py from proj.celery import app app.task def add(x, y): return x y启动一个 Workercelery -A proj worker --loglevelinfo然后在另一个终端发送任务# 1. 基本调用add(2, 2)输出 task id celery -A proj call proj.tasks.add -a [2, 2] # 2. 关键字参数调用add(x1, y2) celery -A proj call proj.tasks.add -k {x: 1, y: 2} # 3. 30 秒后执行 celery -A proj call proj.tasks.add -a [2, 2] --countdown 30 # 4. 指定过期时间60 秒后过期 celery -A proj call proj.tasks.add -a [2, 2] --expires 60 # 5. 路由到指定队列 celery -A proj call proj.tasks.add -a [2, 2] --queue high_priority执行后终端会输出类似d2e0f8f4-...的 UUID这就是任务 ID。你可以用celery result task_id查询执行结果celery -A proj result d2e0f8f4-...注意事项与限制任务未被消费时不产生结果celery call只负责发送消息。如果对应的任务没有 Worker 消费消息会滞留在队列中或在过期后丢弃不会自动执行task_always_eager无效如前所述send_task是纯消息发送路径不会同步执行任务参数必须是合法 JSON--args和--kwargs的解析依赖json.loads非法 JSON 或类型不匹配如--args传了对象会直接报错序列化器需与 Worker 兼容任务消息的序列化器必须与 Worker 端配置一致否则消息可能无法被反序列化敏感信息命令行参数会出现在 Shell 历史记录与进程列表中生产环境请勿通过--kwargs传递敏感数据。总结celery call是 Celery 面向命令行场景的任务发送器其价值在于调试友好无需启动 Python REPL 即可验证任务参数与路由配置脚本化可嵌入 Shell 脚本、CI 流水线实现任务的自动化触发按名称调用基于send_task的能力即使任务未在本地注册也能发送为跨应用、跨进程调试提供便利。理解了 celery/bin/call.py 中call()函数对send_task的透传关系你就掌握了它全部选项的语义——每一个 CLI 选项都对应send_task的一个同名参数最终汇入 celery/app/base.py 的消息发布链路。赞分享任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载相关推荐oh-my-zsh celery 插件Celery 命令行补全机制、子命令参数与源码实现详解oh my zsh celery 插件Celery 命令行补全机制、子命令参数与源码实现详解 oh my zsh 的 celery 插件为 Celery 分布CLI开发工具插件系统SerenityOS pkill 命令完全指南按进程名精确发送信号的原理与实践SerenityOS pkill 命令完全指南按进程名精确发送信号的原理与实践 导读 pkill 是 SerenityOS 中用于 按进程名而非 PID查操作系统内核驱动Celery 3.0Chiastic Slide深度解读无线程 Worker、Canvas 任务链与 celery 伞命令Celery 3.0Chiastic Slide深度解读无线程 Worker、Canvas 任务链与 celery 伞命令 本文基于 Celery 仓库中任务调度后端消息队列上一篇Irodori-TTS-500M-v2安全性指南如何负责任地使用语音克隆技术下一篇如何永久保存微信聊天记录WeChatMsg完全指南与个人数据资产管理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考