任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载Celery 的celery.worker.strategy模块定义了 worker 消费任务消息后的首道处理逻辑——任务执行策略Task Execution Strategy。它是 worker 从 Broker 收到消息到真正进入执行池之间的分发中枢负责解析协议 1/2 消息、构造Request对象、处理 ETA/到期时间、执行速率限制并最终决定任务何时何地进入执行。阅读本文后你将掌握 Celery 消息从 Broker 到执行池的完整调用链、default策略的分支决策逻辑以及 ETA 调度与限流在源码层面的实现原理。模块定位worker 消息处理的优化枢纽celery/worker/strategy.py是 worker 消费侧的核心性能敏感模块。模块顶部注释直接点明其性质Task execution strategy (optimization)任务执行策略属优化代码。这意味着策略层的一切设计都以减少函数调用、缓存属性查找、避免重复计算为第一目标模块源码中的注释也明确标注了这一点见 celery/worker/strategy.py 中 We cache globals and attribute lookups 的说明。整个模块对外只暴露一个符号__all__ (default,)也就是说Celery 的默认任务执行策略是唯一的公开入口default。它通过Task基类上的类属性被引用Task.Strategy celery.worker.strategy:default见 celery/app/task.py 中Strategy属性定义Task.Request celery.worker.request:Request同处定义的请求类策略的实例化入口Task基类提供了start_strategy方法celery/app/task.pydef start_strategy(self, app, consumer, **kwargs): return instantiate(self.Strategy, self, app, consumer, **kwargs)它通过instantiate按字符串形式的路径解析并实例化策略工厂函数。worker 的 Consumer 在启动时调用update_strategies()celery/worker/consumer/consumer.py为应用注册的每一个任务单独创建一份策略实例并缓存进self.strategies字典def update_strategies(self): loader self.app.loader for name, task in self.app.tasks.items(): self.strategies[name] task.start_strategy(self.app, self) task.__trace__ build_tracer(name, task, loader, self.hostname, appself.app)消息到达时的策略查找当新的消息从 Broker 到达时Consumer 的create_task_handler创建的on_task_received回调celery/worker/consumer/consumer.py会按任务名查表分发try: type_ message.headers[task] # protocol v2 except TypeError: return on_unknown_message(None, message) except KeyError: # ... 按 protocol v1 解析 payload 中的 task 字段 try: strategy strategies[type_] except KeyError as exc: return on_unknown_task(None, message, exc)这里清晰地展示了消息协议的分流逻辑协议 v2 消息任务名直接位于message.headers[task]无需解码 body 即可查表协议 v1 消息需要先message.decode()再从payload[task]中获取任务名如果既无法解析消息格式、也查不到对应任务则分别落入on_unknown_message/on_unknown_task兜底处理。查表命中后策略收到(message, payload, ack, reject, callbacks)五个参数——其中ack与reject是延迟执行的 promisecall_soon_ack包装以保证消息确认操作与连接重启恢复 prefetch 计数的逻辑正确协同。default 策略工厂闭包与缓存优化default是一个策略工厂函数strategy factory它接收task、app、consumer三个参数返回一个真正的消息处理函数task_message_handler。这种工厂 闭包设计使每条任务的消息处理路径在 worker 生命周期内只构造一次而所有可复用的对象与属性都在闭包外层一次性取出并局部绑定def default(task, app, consumer, infologger.info, errorlogger.error, task_reservedtask_reserved, to_system_tztimezone.to_system, bytesbytes, proto1_to_proto2proto1_to_proto2):闭包外层缓存的典型对象包括hostname consumer.hostname、connection_errors consumer.connection_errors事件派发器eventer consumer.event_dispatcher及其send方法若未启用则为None定时器call_at consumer.timer.call_at、ETA 执行回调consumer.apply_eta_task限流令牌桶get_bucket consumer.task_buckets.__getitem__、限流函数consumer._limit_task与consumer._limit_post_eta任务请求类Req create_request_cls(Request, task, consumer.pool, hostname, eventer, appapp)。create_request_cls来自 celery/worker/request.py会基于task.Request默认celery.worker.request:Request动态生成一个绑定好任务、连接池、事件派发器等上下文的请求子类。测试 t/unit/worker/test_strategy.py 中的test_custom_request_for_default_strategy验证了自定义Request子类会被正确实例化——这是扩展点之一。注意模块 docstring 与default函数的 docstring 均明确提示由于策略层是优化代码自定义覆盖并不容易its not very easy to override。若确需定制应通过Task.Strategy类属性指向自己的策略工厂并完整复制默认策略的分支逻辑。消息协议归一化proto1 / hybrid 转 proto2Celery 的 Worker 需要同时兼容协议 1 与协议 2 的消息。strategy.py为此提供了两个转换函数将旧协议消息统一归一到协议 2 的(body, headers, decoded, utc)四元组结构。proto1_to_proto2将协议 1 的 bodyargs/kwargs平铺在 payload 中转换为协议 2 结构def proto1_to_proto2(message, body): try: args, kwargs body.get(args, ()), body.get(kwargs, {}) kwargs.items # 校验 kwargs 必须是 mapping except KeyError: raise InvalidTaskError(Message does not have args/kwargs) except AttributeError: raise InvalidTaskError(Task keyword arguments must be a mapping) body.update( argsreprsaferepr(args), kwargsreprsaferepr(kwargs), headersmessage.headers, ) try: body[group] body[taskset] # 旧字段 taskset 映射为 group except KeyError: pass embed { callbacks: body.get(callbacks), errbacks: body.get(errbacks), chord: body.get(chord), chain: None, } return (args, kwargs, embed), body, True, body.get(utc, True)关键细节参数校验args缺失时默认空元组、kwargs缺失时默认空字典但若kwargs不是 mapping如元组立即抛出InvalidTaskError字段归一旧协议字段taskset被映射为group返回值(args, kwargs, embed)作为新 bodybody同时作为 headersdecodedTrue表示参数已解出utc取原消息的utc字段默认True。hybrid_to_proto2处理协议 1/2 混合消息参数位于message.payload但消息头为空或不全。该函数从 payload 中提取协议 2 所需的完整 headers 集合headers { lang: body.get(lang), task: body.get(task), id: body.get(id), root_id: body.get(root_id), parent_id: body.get(parent_id), group: body.get(group), meth: body.get(meth), shadow: body.get(shadow), eta: body.get(eta), expires: body.get(expires), retries: body.get(retries, 0), timelimit: body.get(timelimit, (None, None)), argsrepr: body.get(argsrepr), kwargsrepr: body.get(kwargsrepr), origin: body.get(origin), } headers.update(message.headers or {}) # 以真实消息头覆盖注意retries与timelimit的默认值retries缺省为 0timelimit缺省为(None, None)。测试 t/unit/worker/test_strategy.py 的test_retries_default_value/test_retries_custom_value对此做了专门验证。headers.update(message.headers or {})保证了自定义消息头如测试中的custom: header能够透传。消息归一化的分流逻辑在task_message_handler内部两条转换路径按如下规则选择if body is None and args not in message.payload: # 协议 2 标准消息直接取 message.body / message.headers body, headers, decoded, utc ( message.body, message.headers, False, app.uses_utc_timezone(), ) else: if args in message.payload: # 混合协议payload 中直接携带 args/kwargs body, headers, decoded, utc hybrid_to_proto2(message, message.payload) else: # 协议 1body 中携带 args/kwargs body, headers, decoded, utc proto1_to_proto2(message, body)协议 2 标准消息不预先解码 bodydecodedFalse而是把反序列化工作延迟到执行池中这是协议 2 的核心性能优势之一。utc标志决定后续 ETA 时间戳转换时是否按 UTC 解释否则按app.timezone。task_message_handler完整的任务接收决策流程经过消息归一化后task_message_handler开始执行核心决策流程。它接收(message, body, ack, reject, callbacks)其中callbacks是 Consumer 注册的on_task_message回调链。第一步构造 Request 与接收日志req Req( message, on_ackack, on_rejectreject, appapp, hostnamehostname, eventereventer, tasktask, connection_errorsconnection_errors, bodybody, headersheaders, decodeddecoded, utcutc, )随后在 INFO 级别记录LOG_RECEIVED日志格式串来自celery.app.trace日志上下文包含id、name、args、kwargs、eta五个字段并通过extra{data: context}传给自定义日志处理器结构化日志钩子。第二步撤销revoke检查if (req.expires or req.id in revoked_tasks) and req.revoked(): returnrevoked_tasks consumer.controller.state.revoked是 controller 维护的已撤销任务集合。若任务带有expires或 id 在撤销集合中且req.revoked()判定为真则直接返回——消息既不 ack 也不执行。单元测试test_when_revoked验证了这一短路路径。第三步task_received 信号signals.task_received.send(senderconsumer, requestreq)每个被接收的任务都会触发task_received信号测试test_signal_task_received确认信号参数为senderconsumer, requestreq。第四步task-received 事件若事件系统启用eventer.enabled且任务设置了send_events则发送task-received监控事件send_event( task-received, uuidreq.id, namereq.name, argsreq.argsrepr, kwargsreq.kwargsrepr, root_idreq.root_id, parent_idreq.parent_id, retriesreq.request_dict.get(retries, 0), etareq.eta and req.eta.isoformat(), expiresreq.expires and req.expires.isoformat(), )第五步ETA 时间戳换算if req.eta: try: if req.utc: eta to_timestamp(to_system_tz(req.eta)) else: eta to_timestamp(req.eta, app.timezone) except (OverflowError, ValueError) as exc: error(Couldnt convert ETA %r to timestamp: %r. Task: %r, ...) req.reject(requeueFalse) # ETA 非法直接拒绝且不重新入队to_timestamp来自kombu.asynchronous.timer将 datetime 换算为单调时钟时间戳。换算失败越界或格式非法时记录错误日志并reject(requeueFalse)——注意这会导致消息被丢弃。第六步四路分支决策这是整个策略的核心决策点依据eta定时与bucket限流令牌桶两个布尔条件组合出四种执行路径if rate_limits_enabled: bucket get_bucket(task.name) # 从 task_buckets 按任务名取令牌桶 if eta and bucket: # 路径 A既有 ETA 又有限流 consumer.qos.increment_eventually() req._eta_timer_entry call_at( eta, limit_post_eta, (req, bucket, 1), priority6) task_scheduled(req) return if eta: # 路径 B仅 ETA consumer.qos.increment_eventually() req._eta_timer_entry call_at( eta, apply_eta_task, (req,), priority6) task_scheduled(req) return task_message_handler if bucket: # 路径 C仅限流 return limit_task(req, bucket, 1) # 路径 D立即执行 task_reserved(req) if callbacks: [callback(req) for callback in callbacks] handle(req)各路径语义如下路径触发条件处理方式定时器回调AETA 限流同时命中定时到点后先进限流桶排队consumer._limit_post_eta先qos.decrement_eventually()再入桶B仅 ETA定时到点后立即执行consumer.apply_eta_tasktask_reserved→on_task_request→qos.decrement_eventually()C仅限流直接进入限流桶等待放行consumer._limit_taskbucket.add((request, tokens))后按桶调度D无条件立即标记 reserved 并交给on_task_request无限流实现细节celery/worker/consumer/consumer.py_limit_task将请求加入kombu.utils.limits.TokenBucket令牌桶然后通过_schedule_bucket_request按bucket.expected_time(tokens)计算等待时间用self.timer.call_after调度到点取出请求。路径 A 之所以需要_limit_post_eta是因为它在 ETA 定时到点后才入桶而 ETA 等待期间 QoS prefetch 计数已经递增过因此到点入桶前需要qos.decrement_eventually()归还计数——这是路径 A 与路径 C 的唯一区别C 是立即入桶无待还计数。测试类test_default_strategy_proto2t/unit/worker/test_strategy.py中的was_reserved/was_rate_limited/was_limited_with_eta/was_scheduled断言精确对应了这四条路径。ETA 任务的注册时机与并发安全路径 A/B 中有一个非常关键的顺序先挂载定时器条目再调用task_scheduled(req)req._eta_timer_entry call_at(eta, ..., priority6) task_scheduled(req)Request._eta_timer_entry属性celery/worker/request.py 中定义保存定时器条目其 docstring 说明了用途连接丢失时Consumer.on_close需要取消该请求挂起的 ETA 回调。task_scheduledcelery/worker/state.py将请求注册进 worker 的全局state.requests/state.scheduled_requests使query_task等远程控制命令能查询到尚未到点执行的 ETA 任务。这两个操作的先后顺序有严格的并发语义源码注释解释了原因对非事件循环的进程池timer2.Timer运行在独立线程上on_close()可能并发触发若先注册再挂定时器on_close可能观察到已注册但无定时器条目可取消的半注册状态导致定时器条目泄漏。因此必须先call_at后task_scheduled。对应的两条回归测试test_eta_task_registers_request_in_state验证 ETA 任务在到点前即可通过state.requests查到回归 #5321test_eta_task_timer_entry_attached_before_scheduled_visible验证task_scheduled被调用时_eta_timer_entry已非空。扩展点自定义 Request 与自定义策略自定义 Request 类Task.Request类属性允许为单个任务指定自定义请求类。策略通过symbol_by_name(task.Request)解析该类再由create_request_cls绑定上下文。测试test_custom_request_for_default_strategy展示了最小用法class MyRequest(Request): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 自定义逻辑 class MyTask(Task): Request MyRequest app.task(baseMyTask) def my_task(): ...自定义策略Task.Strategy类属性默认celery.worker.strategy:default指定策略工厂。自定义策略必须接受(task, app, consumer)并返回task_message_handler形式的函数。由于默认策略闭包内缓存了大量 consumer 属性自定义策略通常需要以默认实现为模板在其基础上调整特定分支逻辑如自定义 ETA 处理或限流行为。完整调用链总览将以上各环节串联一条任务消息在 worker 内的完整旅程为Broker 消息 → Consumer.create_task_handler().on_task_received [按 headers[task] 查 strategies 表] → task_message_handlerdefault 策略返回的处理函数 ├─ 协议归一化proto2 直接 / hybrid_to_proto2 / proto1_to_proto2 ├─ 构造 Requestcreate_request_cls ├─ 接收日志LOG_RECEIVED task_received 信号 task-received 事件 ├─ ETA 换算 撤销检查 └─ 四路分支 ├─ ETA限流 → timer → _limit_post_eta → 令牌桶 → 执行 ├─ ETA → timer → apply_eta_task → task_reserved → on_task_request ├─ 限流 → _limit_task → 令牌桶 → 执行 └─ 立即 → task_reserved → on_task_request → pool 执行 → trace / 结果后端关键实现文件索引策略实现celery/worker/strategy.py策略注册与消息分发celery/worker/consumer/consumer.pyupdate_strategies/create_task_handler策略与请求类默认值celery/app/task.pyTask.Strategy/Task.Request/start_strategyRequest 实现与 ETA 定时器条目celery/worker/request.py全局任务状态注册celery/worker/state.pytask_reserved/task_scheduled策略单元测试t/unit/worker/test_strategy.py覆盖协议转换、四路分支、信号、撤销、ETA 并发回归等全部路径总结celery.worker.strategy是 Celery Worker 消费链路上的性能要害它以工厂闭包 属性缓存的优化风格把每条任务的消息接收路径压缩到最小开销以协议归一化函数兼容新旧两种消息协议并以eta与限流令牌桶两个条件组合出四条精确的执行分支。理解这一模块等于掌握了 Celery 从消息到达到进入执行池之间全部的分发、调度与限流语义——无论是排查任务延迟、定制 ETA 行为还是扩展自定义 Request都需要从这里入手。赞分享任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载相关推荐英雄联盟录像制作全攻略用League Director打造专业游戏视频英雄联盟录像制作全攻略用League Director打造专业游戏视频 League Director是一款由Riot Games官方开发的免费开源工具专为任务调度后端消息队列Nacos 任务执行引擎Task Execution深度解析延迟任务、执行任务与领域调度机制Nacos 任务执行引擎Task Execution深度解析延迟任务、执行任务与领域调度机制 本文以 Nacos 官方设计规范 Foundation Ta后端微服务配置中心服务注册发现云原生Onyx 后台任务架构深度解析Celery Worker、队列映射与调度机制全指南Onyx 后台任务架构深度解析Celery Worker、队列映射与调度机制全指南 本文以 Onyx开源 AI 平台的后台任务体系为核心系统讲解其基于AI 应用大模型RAGAI Agent后端前端上一篇symbols-outline.nvim快捷键全解析提升代码浏览效率的终极指南下一篇10分钟快速上手如何用RVC WebUI打造专业级AI语音转换系统创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考