Celery Worker 心跳机制深度解析celery.worker.consumer.heart模块与事件心跳实现【免费下载链接】celeryDistributed Task Queue (development branch)项目地址: https://gitcode.com/gh_mirrors/ce/celery导读本文围绕 Celery 分布式任务队列中 Worker 的**事件心跳Event Heartbeat**机制展开以 celery.worker.consumer.heart 模块为核心剖析Heartbootstep 如何周期性地向消息代理发送worker-heartbeat事件、如何与Eventsbootstep 协作、如何通过heartbeat_sent信号与heartbeat_interval/without_heartbeat等配置协同工作并深入 celery.worker.heartbeat 源码与对应单元测试帮助读者理解 Worker 心跳的完整生命周期、故障检测价值以及如何在监控系统中正确使用心跳数据。适用场景当你需要理解 Celery Worker 如何被监控系统如 Flower、celery events发现与跟踪、排查Worker 掉线但监控无感知类问题或自定义心跳频率与事件内容时本文提供从配置到源码的完整路径。一、模块定位Worker 消费者蓝图中的心跳 BootstepCelery Worker 的启动由多个 bootstep 组成每个 bootstep 负责一个独立的关注点。Heart是 Worker 消费者Consumer蓝图中的一个StartStopStep其核心职责是每 N 秒发送一次worker-heartbeat事件。该模块定义在 celery/worker/consumer/heart.pyWorker Event Heartbeat Bootstep. from celery import bootsteps from celery.worker import heartbeat from .events import Events __all__ (Heart,) class Heart(bootsteps.StartStopStep): Bootstep sending event heartbeats. This service sends a worker-heartbeat message every n seconds. Note: Not to be confused with AMQP protocol level heartbeats. requires (Events,) def __init__(self, c, without_heartbeatFalse, heartbeat_intervalNone, **kwargs): self.enabled not without_heartbeat self.heartbeat_interval heartbeat_interval c.heart None super().__init__(c, **kwargs) def start(self, c): c.heart heartbeat.Heart( c.timer, c.event_dispatcher, self.heartbeat_interval, ) c.heart.start() def stop(self, c): c.heart c.heart and c.heart.stop() shutdown stop1.1 与 AMQP 协议级心跳的严格区分模块文档字符串中特别强调Not to be confused with AMQP protocol level heartbeats.Celery 中存在两类完全不同的心跳维度事件心跳Event HeartbeatAMQP 协议级心跳实现位置celery.worker.consumer.heart:HeartKombu 传输层 /broker_heartbeat配置载体worker-heartbeat事件消息AMQP 协议帧用途供监控系统追踪 Worker 存活与负载维持 TCP 连接活性、检测 broker 断连周期默认值2 秒事件发送间隔120 秒broker_heartbeatAMQP 协议级心跳由broker_heartbeat默认 120 秒与broker_heartbeat_checkrate默认 3.0配置控制见 celery/app/defaults.pyheartbeatOption(120, typeint), heartbeat_checkrateOption(3.0, typeint),而本文讨论的事件心跳是 Worker 应用层主动发布到事件流中的业务事件二者互不替代。1.2 Bootstep 的依赖关系与蓝图注册Heart通过requires (Events,)声明依赖即必须先启动事件分发器Events bootstep才能发送心跳事件。Eventsbootstep 又依赖Connection形成完整的依赖链Connection → Events → HeartConsumer 蓝图中将Heart注册为默认步骤之一见 celery/worker/consumer/consumer.pydefault_steps [ celery.worker.consumer.connection:Connection, celery.worker.consumer.mingle:Mingle, celery.worker.consumer.events:Events, celery.worker.consumer.gossip:Gossip, celery.worker.consumer.heart:Heart, celery.worker.consumer.control:Control, celery.worker.consumer.tasks:Tasks, celery.worker.consumer.delayed_delivery:DelayedDelivery, celery.worker.consumer.consumer:Evloop, celery.worker.consumer.agent:Agent, ]由于Heart依赖Events即使Heart本身被禁用只要Events保持启用例如--without-gossip场景下Events.send_events的计算逻辑事件系统仍会工作但不会再有周期性心跳事件。二、初始化逻辑enabled 开关与 interval 透传Heart.__init__接受两个关键参数分别对应两条命令行选项def __init__(self, c, without_heartbeatFalse, heartbeat_intervalNone, **kwargs): self.enabled not without_heartbeat self.heartbeat_interval heartbeat_interval c.heart None参数来源作用without_heartbeat--without-heartbeat命令行选项为True时禁用该 bootstepenabled Falseheartbeat_interval--heartbeat-interval命令行选项透传给底层Heart服务控制发送周期秒两个 CLI 选项定义在 celery/bin/worker.pyclick.option(--without-heartbeat, is_flagTrue, clsCeleryOption, help_groupFeatures, ) click.option(--heartbeat-interval, typeint, clsCeleryOption, help_groupFeatures, )使用示例# 关闭事件心跳监控将无法感知 Worker 在线状态 celery -A proj worker --without-heartbeat # 将心跳周期从默认 2 秒调整为 20 秒降低事件频率监控灵敏度随之降低 celery -A proj worker --heartbeat-interval 20注意--without-heartbeat是is_flagTrue的开关选项而--heartbeat-interval接受int类型参数。若仅传入heartbeat_interval而不传without_heartbeatenabled仍为True只是发送周期被调整。相关配置项send_task_events旧名celery_send_events默认False控制是否发送任务事件定义于 celery/app/defaults.py。任务事件与心跳事件相互独立——心跳由Heartbootstep 负责任务事件由Eventsbootstep 的groups过滤。三、start/stop心跳服务的生命周期管理作为StartStopStepHeart将生命周期管理委托给真正的服务类celery.worker.heartbeat.Heartdef start(self, c): c.heart heartbeat.Heart( c.timer, c.event_dispatcher, self.heartbeat_interval, ) c.heart.start() def stop(self, c): c.heart c.heart and c.heart.stop() shutdown stopstart(c)用 Consumer 的timer高优先级内部定时器与event_dispatcher事件分发器由Eventsbootstep 创建实例化心跳服务并立即启动。stop(c)幂等停止——若c.heart为None从未启动表达式结果为None不会抛错。shutdown stop优雅关闭与强制关闭共用同一实现。c.timer在 celery/worker/consumer/consumer.py 中被注释为用于高优先级内部任务例如发送心跳的定时器印证了心跳对实时性的要求。四、底层实现celery.worker.heartbeat.Heart服务真正执行心跳发送的是 celery/worker/heartbeat.py 中的Heart类class Heart: Timer sending heartbeats at regular intervals. Arguments: timer (kombu.asynchronous.timer.Timer): Timer to use. eventer (celery.events.EventDispatcher): Event dispatcher to use. interval (float): Time in seconds between sending heartbeats. Default is 2 seconds. def __init__(self, timer, eventer, intervalNone): self.timer timer self.eventer eventer self.interval float(interval or 2.0) self.tref None # Make event dispatcher start/stop us when enabled/disabled. self.eventer.on_enabled.add(self.start) self.eventer.on_disabled.add(self.stop) # Only send heartbeat_sent signal if it has receivers. self._send_sent_signal ( heartbeat_sent.send if heartbeat_sent.receivers else None) def _send(self, event, retryTrue): if self._send_sent_signal is not None: self._send_sent_signal(senderself) return self.eventer.send(event, freqself.interval, activelen(active_requests), processedall_total_count[0], loadavgload_average(), retryretry, **SOFTWARE_INFO) def start(self): if self.eventer.enabled: self._send(worker-online) self.tref self.timer.call_repeatedly( self.interval, self._send, (worker-heartbeat,), ) def stop(self): if self.tref is not None: self.timer.cancel(self.tref) self.tref None if self.eventer.enabled: self._send(worker-offline, retryFalse)4.1 默认间隔与关键字段interval默认2.0秒float(interval or 2.0)即每 2 秒发送一次worker-heartbeat事件。tref保存call_repeatedly返回的定时器句柄用于stop()时取消。eventer.on_enabled/eventer.on_disabled注册为回调事件分发器启用/禁用时会自动启动/停止心跳服务实现分发器驱动心跳的联动。4.2 每条心跳事件携带的负载_send方法通过eventer.send发布事件每一条worker-heartbeat事件包含以下字段字段来源含义freqself.interval心跳发送频率秒供监控方预估下一跳时间activelen(active_requests)当前活跃请求数来自 celery/worker/state.pyprocessedall_total_count[0]已处理任务总数loadavgload_average()系统负载平均值来自 celery/utils/sysinfo.py**SOFTWARE_INFOcelery.worker.state.SOFTWARE_INFOCelery 版本等软件信息4.3 在线/离线事件的完整生命周期start()与stop()揭示了 Worker 生命周期中的三类事件def start(self): if self.eventer.enabled: self._send(worker-online) # ① 上线事件仅一次 self.tref self.timer.call_repeatedly( self.interval, self._send, (worker-heartbeat,), # ② 周期心跳 ) def stop(self): if self.tref is not None: self.timer.cancel(self.tref) self.tref None if self.eventer.enabled: self._send(worker-offline, retryFalse) # ③ 下线事件仅一次不重试worker-online心跳服务启动时发送一次宣告 Worker 上线。worker-heartbeat每interval秒周期发送是监控系统判断 Worker 存活的核心信号。worker-offline停止时发送一次且retryFalse下线事件不重试因为连接可能已不可用。4.4heartbeat_sent信号celery/signals.py 定义了heartbeat_sent信号且Heart.__init__中做了一个性能优化——仅当信号存在接收者时才发送# Only send heartbeat_sent signal if it has receivers. self._send_sent_signal ( heartbeat_sent.send if heartbeat_sent.receivers else None)这意味着没有订阅者时心跳发送路径不承担信号分发开销。订阅示例from celery.signals import heartbeat_sent heartbeat_sent.connect def on_heartbeat(sender, **kwargs): print(fHeartbeat sent by {sender!r})注意_send中信号发送先于eventer.send即先通知本进程内的订阅者再真正发布事件到 broker。五、与 Events bootstep 的协作关系Heart依赖Events而Events定义于 celery/worker/consumer/events.py负责创建事件分发器class Events(bootsteps.StartStopStep): Service used for sending monitoring events. requires (Connection,) def __init__(self, c, task_eventsTrue, without_heartbeatFalse, without_gossipFalse, **kwargs): self.groups None if task_events else [worker] self.send_events ( task_events or not without_gossip or not without_heartbeat ) self.enabled self.send_events c.event_dispatcher None关键逻辑在send_events的计算只要任务事件、gossip 或 heartbeat 任一开启事件分发器就必须启用。这保证--without-heartbeat不会连带关闭整个事件系统Events仍需为 gossip 等服务工作。Events.start创建分发器时还做了连接层的心跳配合def start(self, c): conn c.connection_for_write(heartbeatc.amqheartbeat) dis c.event_dispatcher c.app.events.Dispatcher( conn, hostnamec.hostname, enabledself.send_events, groupsself.groups, buffer_group[task] if c.hub else None, on_send_bufferedc.on_send_event_buffered if c.hub else None, ) # register for reads so broker heartbeats are consumed. if c.hub and c.amqheartbeat and conn.supports_heartbeats: conn.transport.register_with_event_loop(conn.connection, c.hub)c.amqheartbeat的取值逻辑在 celery/worker/consumer/consumer.pyself.hub hub if self.hub or getattr(self.pool, is_green, False): self.amqheartbeat amqheartbeat if self.amqheartbeat is None: self.amqheartbeat self.app.conf.broker_heartbeat else: self.amqheartbeat 0即异步事件循环hub或绿色线程池eventlet/gevent环境下AMQP 心跳才启用默认值取broker_heartbeat120 秒同步循环下amqheartbeat为 0禁用。事件分发器发送心跳所用的连接会携带该 AMQP 心跳参数使事件通道同时具备连接保活能力。六、源码级验证单元测试如何锁定行为仓库单元测试对Heartbootstep 与底层服务的契约有明确断言可作为理解与回归的参考。6.1 Bootstep 层测试t/unit/worker/test_consumer.py 中test_Heart测试类验证class test_Heart: def test_start(self): c Mock() c.timer Mock() c.event_dispatcher Mock() with patch(celery.worker.heartbeat.Heart) as hcls: h Heart(c) assert h.enabled # 默认启用 assert h.heartbeat_interval is None # 默认间隔为空走 2 秒默认值 assert c.heart is None # 未启动前 c.heart 为 None h.start(c) assert c.heart # start 后创建服务 hcls.assert_called_with(c.timer, c.event_dispatcher, h.heartbeat_interval) c.heart.start.assert_called_with() # 服务被启动 def test_start_heartbeat_interval(self): c Mock() c.timer Mock() c.event_dispatcher Mock() with patch(celery.worker.heartbeat.Heart) as hcls: h Heart(c, False, 20) # without_heartbeatFalse, interval20 assert h.enabled assert h.heartbeat_interval 20 # 间隔被透传 ...测试确认了三个契约Heart(c)默认enabledTrue、heartbeat_intervalNone、c.heart初始为Nonestart(c)使用c.timer、c.event_dispatcher与heartbeat_interval实例化底层服务并调用其start()显式传入heartbeat_interval20会被完整透传。6.2 服务层测试t/unit/worker/test_heartbeat.py 针对celery.worker.heartbeat.Heart验证了事件序列节选关键断言from celery.worker.heartbeat import Heart def test_start_sends_online_and_schedules_heartbeats(self): h Heart(timer, eventer, interval1) ... # Invoke a heartbeat assert eventer.sent[-1][0] worker-heartbeat结合 t/unit/worker/test_consumer.py 中dispatcher.send(worker-heartbeat, freq5)的模拟以及 t/unit/worker/test_control.py 中panel.handle(heartbeat)触发(worker-heartbeat,)事件发送的断言可以确认心跳事件的类型名恒为worker-heartbeat且freq字段携带发送周期。七、心跳在监控生态中的角色与 FAQ7.1 监控系统如何利用心跳Flower、celery events、celery events --camera等监控工具通过监听事件流中的worker-online/worker-heartbeat/worker-offline事件来构建 Worker 在线状态视图。判断逻辑通常为收到worker-online→ Worker 标记为在线持续收到worker-heartbeat→ Worker 存活且可从事件字段读取active活跃请求数、processed处理任务数、loadavg系统负载用于负载展示超过若干个freq周期未收到心跳 → Worker 被判定为离线或失联。因此--heartbeat-interval的大小直接决定监控系统的故障发现延迟与事件流量之间的权衡。7.2 常见问题Q1--without-heartbeat与send_task_events是什么关系二者独立。--without-heartbeat只关闭心跳 bootstepHeart.enabled Falsesend_task_events旧名celery_send_events控制是否发送任务级事件由Eventsbootstep 的groups过滤实现。关闭心跳不会自动关闭任务事件反之亦然。Q2为什么建议不要在--without-heartbeat下依赖监控心跳是监控系统感知 Worker 存活状态的主要信号源在 gossip 关闭时几乎是唯一来源。Events.__init__中send_events的计算保证了 heartbeat 关闭时事件分发器仍可用但不会再产生周期性心跳监控端将因缺少心跳事件而无法及时感知 Worker 掉线。Q3心跳事件发送失败会怎样_send默认retryTrueworker-offline除外发送失败会由事件分发器按既有重试策略处理worker-offline使用retryFalse避免在连接已不可用时做无谓重试。Q4事件心跳能用于判断任务正在运行吗可以间接判断。worker-heartbeat事件携带active当前活跃请求数字段监控端可据此感知 Worker 是否正在处理任务但它不是任务粒度的事件——任务粒度应依赖task-started/task-succeeded等任务事件需开启send_task_events。八、小结从配置到源码的完整链路层级文件职责CLI 选项celery/bin/worker.py--without-heartbeat、--heartbeat-intervalBootstepcelery/worker/consumer/heart.pyHeartenabled 开关、依赖Events、生命周期委托服务实现celery/worker/heartbeat.py周期发送worker-heartbeat附带active/processed/loadavg事件分发celery/worker/consumer/events.py创建EventDispatcher联动 AMQP 心跳配置默认值celery/app/defaults.pybroker_heartbeat120、broker_heartbeat_checkrate3.0AMQP 级信号钩子celery/signals.pyheartbeat_sent有接收者时才分发测试印证t/unit/worker/test_consumer.py、t/unit/worker/test_heartbeat.py锁定 bootstep 契约与事件序列理解这条链路后你可以精准地控制 Worker 心跳行为用--heartbeat-interval调节故障发现灵敏度用--without-heartbeat在不需要监控时降低事件开销通过heartbeat_sent信号在本进程内挂钩子或直接消费worker-heartbeat事件构建自定义的 Worker 监控大盘。【免费下载链接】celeryDistributed Task Queue (development branch)项目地址: https://gitcode.com/gh_mirrors/ce/celery创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Celery Worker 心跳机制深度解析：`celery.worker.consumer.heart` 模块与事件心跳实现
Celery Worker 心跳机制深度解析celery.worker.consumer.heart模块与事件心跳实现【免费下载链接】celeryDistributed Task Queue (development branch)项目地址: https://gitcode.com/gh_mirrors/ce/celery导读本文围绕 Celery 分布式任务队列中 Worker 的**事件心跳Event Heartbeat**机制展开以 celery.worker.consumer.heart 模块为核心剖析Heartbootstep 如何周期性地向消息代理发送worker-heartbeat事件、如何与Eventsbootstep 协作、如何通过heartbeat_sent信号与heartbeat_interval/without_heartbeat等配置协同工作并深入 celery.worker.heartbeat 源码与对应单元测试帮助读者理解 Worker 心跳的完整生命周期、故障检测价值以及如何在监控系统中正确使用心跳数据。适用场景当你需要理解 Celery Worker 如何被监控系统如 Flower、celery events发现与跟踪、排查Worker 掉线但监控无感知类问题或自定义心跳频率与事件内容时本文提供从配置到源码的完整路径。一、模块定位Worker 消费者蓝图中的心跳 BootstepCelery Worker 的启动由多个 bootstep 组成每个 bootstep 负责一个独立的关注点。Heart是 Worker 消费者Consumer蓝图中的一个StartStopStep其核心职责是每 N 秒发送一次worker-heartbeat事件。该模块定义在 celery/worker/consumer/heart.pyWorker Event Heartbeat Bootstep. from celery import bootsteps from celery.worker import heartbeat from .events import Events __all__ (Heart,) class Heart(bootsteps.StartStopStep): Bootstep sending event heartbeats. This service sends a worker-heartbeat message every n seconds. Note: Not to be confused with AMQP protocol level heartbeats. requires (Events,) def __init__(self, c, without_heartbeatFalse, heartbeat_intervalNone, **kwargs): self.enabled not without_heartbeat self.heartbeat_interval heartbeat_interval c.heart None super().__init__(c, **kwargs) def start(self, c): c.heart heartbeat.Heart( c.timer, c.event_dispatcher, self.heartbeat_interval, ) c.heart.start() def stop(self, c): c.heart c.heart and c.heart.stop() shutdown stop1.1 与 AMQP 协议级心跳的严格区分模块文档字符串中特别强调Not to be confused with AMQP protocol level heartbeats.Celery 中存在两类完全不同的心跳维度事件心跳Event HeartbeatAMQP 协议级心跳实现位置celery.worker.consumer.heart:HeartKombu 传输层 /broker_heartbeat配置载体worker-heartbeat事件消息AMQP 协议帧用途供监控系统追踪 Worker 存活与负载维持 TCP 连接活性、检测 broker 断连周期默认值2 秒事件发送间隔120 秒broker_heartbeatAMQP 协议级心跳由broker_heartbeat默认 120 秒与broker_heartbeat_checkrate默认 3.0配置控制见 celery/app/defaults.pyheartbeatOption(120, typeint), heartbeat_checkrateOption(3.0, typeint),而本文讨论的事件心跳是 Worker 应用层主动发布到事件流中的业务事件二者互不替代。1.2 Bootstep 的依赖关系与蓝图注册Heart通过requires (Events,)声明依赖即必须先启动事件分发器Events bootstep才能发送心跳事件。Eventsbootstep 又依赖Connection形成完整的依赖链Connection → Events → HeartConsumer 蓝图中将Heart注册为默认步骤之一见 celery/worker/consumer/consumer.pydefault_steps [ celery.worker.consumer.connection:Connection, celery.worker.consumer.mingle:Mingle, celery.worker.consumer.events:Events, celery.worker.consumer.gossip:Gossip, celery.worker.consumer.heart:Heart, celery.worker.consumer.control:Control, celery.worker.consumer.tasks:Tasks, celery.worker.consumer.delayed_delivery:DelayedDelivery, celery.worker.consumer.consumer:Evloop, celery.worker.consumer.agent:Agent, ]由于Heart依赖Events即使Heart本身被禁用只要Events保持启用例如--without-gossip场景下Events.send_events的计算逻辑事件系统仍会工作但不会再有周期性心跳事件。二、初始化逻辑enabled 开关与 interval 透传Heart.__init__接受两个关键参数分别对应两条命令行选项def __init__(self, c, without_heartbeatFalse, heartbeat_intervalNone, **kwargs): self.enabled not without_heartbeat self.heartbeat_interval heartbeat_interval c.heart None参数来源作用without_heartbeat--without-heartbeat命令行选项为True时禁用该 bootstepenabled Falseheartbeat_interval--heartbeat-interval命令行选项透传给底层Heart服务控制发送周期秒两个 CLI 选项定义在 celery/bin/worker.pyclick.option(--without-heartbeat, is_flagTrue, clsCeleryOption, help_groupFeatures, ) click.option(--heartbeat-interval, typeint, clsCeleryOption, help_groupFeatures, )使用示例# 关闭事件心跳监控将无法感知 Worker 在线状态 celery -A proj worker --without-heartbeat # 将心跳周期从默认 2 秒调整为 20 秒降低事件频率监控灵敏度随之降低 celery -A proj worker --heartbeat-interval 20注意--without-heartbeat是is_flagTrue的开关选项而--heartbeat-interval接受int类型参数。若仅传入heartbeat_interval而不传without_heartbeatenabled仍为True只是发送周期被调整。相关配置项send_task_events旧名celery_send_events默认False控制是否发送任务事件定义于 celery/app/defaults.py。任务事件与心跳事件相互独立——心跳由Heartbootstep 负责任务事件由Eventsbootstep 的groups过滤。三、start/stop心跳服务的生命周期管理作为StartStopStepHeart将生命周期管理委托给真正的服务类celery.worker.heartbeat.Heartdef start(self, c): c.heart heartbeat.Heart( c.timer, c.event_dispatcher, self.heartbeat_interval, ) c.heart.start() def stop(self, c): c.heart c.heart and c.heart.stop() shutdown stopstart(c)用 Consumer 的timer高优先级内部定时器与event_dispatcher事件分发器由Eventsbootstep 创建实例化心跳服务并立即启动。stop(c)幂等停止——若c.heart为None从未启动表达式结果为None不会抛错。shutdown stop优雅关闭与强制关闭共用同一实现。c.timer在 celery/worker/consumer/consumer.py 中被注释为用于高优先级内部任务例如发送心跳的定时器印证了心跳对实时性的要求。四、底层实现celery.worker.heartbeat.Heart服务真正执行心跳发送的是 celery/worker/heartbeat.py 中的Heart类class Heart: Timer sending heartbeats at regular intervals. Arguments: timer (kombu.asynchronous.timer.Timer): Timer to use. eventer (celery.events.EventDispatcher): Event dispatcher to use. interval (float): Time in seconds between sending heartbeats. Default is 2 seconds. def __init__(self, timer, eventer, intervalNone): self.timer timer self.eventer eventer self.interval float(interval or 2.0) self.tref None # Make event dispatcher start/stop us when enabled/disabled. self.eventer.on_enabled.add(self.start) self.eventer.on_disabled.add(self.stop) # Only send heartbeat_sent signal if it has receivers. self._send_sent_signal ( heartbeat_sent.send if heartbeat_sent.receivers else None) def _send(self, event, retryTrue): if self._send_sent_signal is not None: self._send_sent_signal(senderself) return self.eventer.send(event, freqself.interval, activelen(active_requests), processedall_total_count[0], loadavgload_average(), retryretry, **SOFTWARE_INFO) def start(self): if self.eventer.enabled: self._send(worker-online) self.tref self.timer.call_repeatedly( self.interval, self._send, (worker-heartbeat,), ) def stop(self): if self.tref is not None: self.timer.cancel(self.tref) self.tref None if self.eventer.enabled: self._send(worker-offline, retryFalse)4.1 默认间隔与关键字段interval默认2.0秒float(interval or 2.0)即每 2 秒发送一次worker-heartbeat事件。tref保存call_repeatedly返回的定时器句柄用于stop()时取消。eventer.on_enabled/eventer.on_disabled注册为回调事件分发器启用/禁用时会自动启动/停止心跳服务实现分发器驱动心跳的联动。4.2 每条心跳事件携带的负载_send方法通过eventer.send发布事件每一条worker-heartbeat事件包含以下字段字段来源含义freqself.interval心跳发送频率秒供监控方预估下一跳时间activelen(active_requests)当前活跃请求数来自 celery/worker/state.pyprocessedall_total_count[0]已处理任务总数loadavgload_average()系统负载平均值来自 celery/utils/sysinfo.py**SOFTWARE_INFOcelery.worker.state.SOFTWARE_INFOCelery 版本等软件信息4.3 在线/离线事件的完整生命周期start()与stop()揭示了 Worker 生命周期中的三类事件def start(self): if self.eventer.enabled: self._send(worker-online) # ① 上线事件仅一次 self.tref self.timer.call_repeatedly( self.interval, self._send, (worker-heartbeat,), # ② 周期心跳 ) def stop(self): if self.tref is not None: self.timer.cancel(self.tref) self.tref None if self.eventer.enabled: self._send(worker-offline, retryFalse) # ③ 下线事件仅一次不重试worker-online心跳服务启动时发送一次宣告 Worker 上线。worker-heartbeat每interval秒周期发送是监控系统判断 Worker 存活的核心信号。worker-offline停止时发送一次且retryFalse下线事件不重试因为连接可能已不可用。4.4heartbeat_sent信号celery/signals.py 定义了heartbeat_sent信号且Heart.__init__中做了一个性能优化——仅当信号存在接收者时才发送# Only send heartbeat_sent signal if it has receivers. self._send_sent_signal ( heartbeat_sent.send if heartbeat_sent.receivers else None)这意味着没有订阅者时心跳发送路径不承担信号分发开销。订阅示例from celery.signals import heartbeat_sent heartbeat_sent.connect def on_heartbeat(sender, **kwargs): print(fHeartbeat sent by {sender!r})注意_send中信号发送先于eventer.send即先通知本进程内的订阅者再真正发布事件到 broker。五、与 Events bootstep 的协作关系Heart依赖Events而Events定义于 celery/worker/consumer/events.py负责创建事件分发器class Events(bootsteps.StartStopStep): Service used for sending monitoring events. requires (Connection,) def __init__(self, c, task_eventsTrue, without_heartbeatFalse, without_gossipFalse, **kwargs): self.groups None if task_events else [worker] self.send_events ( task_events or not without_gossip or not without_heartbeat ) self.enabled self.send_events c.event_dispatcher None关键逻辑在send_events的计算只要任务事件、gossip 或 heartbeat 任一开启事件分发器就必须启用。这保证--without-heartbeat不会连带关闭整个事件系统Events仍需为 gossip 等服务工作。Events.start创建分发器时还做了连接层的心跳配合def start(self, c): conn c.connection_for_write(heartbeatc.amqheartbeat) dis c.event_dispatcher c.app.events.Dispatcher( conn, hostnamec.hostname, enabledself.send_events, groupsself.groups, buffer_group[task] if c.hub else None, on_send_bufferedc.on_send_event_buffered if c.hub else None, ) # register for reads so broker heartbeats are consumed. if c.hub and c.amqheartbeat and conn.supports_heartbeats: conn.transport.register_with_event_loop(conn.connection, c.hub)c.amqheartbeat的取值逻辑在 celery/worker/consumer/consumer.pyself.hub hub if self.hub or getattr(self.pool, is_green, False): self.amqheartbeat amqheartbeat if self.amqheartbeat is None: self.amqheartbeat self.app.conf.broker_heartbeat else: self.amqheartbeat 0即异步事件循环hub或绿色线程池eventlet/gevent环境下AMQP 心跳才启用默认值取broker_heartbeat120 秒同步循环下amqheartbeat为 0禁用。事件分发器发送心跳所用的连接会携带该 AMQP 心跳参数使事件通道同时具备连接保活能力。六、源码级验证单元测试如何锁定行为仓库单元测试对Heartbootstep 与底层服务的契约有明确断言可作为理解与回归的参考。6.1 Bootstep 层测试t/unit/worker/test_consumer.py 中test_Heart测试类验证class test_Heart: def test_start(self): c Mock() c.timer Mock() c.event_dispatcher Mock() with patch(celery.worker.heartbeat.Heart) as hcls: h Heart(c) assert h.enabled # 默认启用 assert h.heartbeat_interval is None # 默认间隔为空走 2 秒默认值 assert c.heart is None # 未启动前 c.heart 为 None h.start(c) assert c.heart # start 后创建服务 hcls.assert_called_with(c.timer, c.event_dispatcher, h.heartbeat_interval) c.heart.start.assert_called_with() # 服务被启动 def test_start_heartbeat_interval(self): c Mock() c.timer Mock() c.event_dispatcher Mock() with patch(celery.worker.heartbeat.Heart) as hcls: h Heart(c, False, 20) # without_heartbeatFalse, interval20 assert h.enabled assert h.heartbeat_interval 20 # 间隔被透传 ...测试确认了三个契约Heart(c)默认enabledTrue、heartbeat_intervalNone、c.heart初始为Nonestart(c)使用c.timer、c.event_dispatcher与heartbeat_interval实例化底层服务并调用其start()显式传入heartbeat_interval20会被完整透传。6.2 服务层测试t/unit/worker/test_heartbeat.py 针对celery.worker.heartbeat.Heart验证了事件序列节选关键断言from celery.worker.heartbeat import Heart def test_start_sends_online_and_schedules_heartbeats(self): h Heart(timer, eventer, interval1) ... # Invoke a heartbeat assert eventer.sent[-1][0] worker-heartbeat结合 t/unit/worker/test_consumer.py 中dispatcher.send(worker-heartbeat, freq5)的模拟以及 t/unit/worker/test_control.py 中panel.handle(heartbeat)触发(worker-heartbeat,)事件发送的断言可以确认心跳事件的类型名恒为worker-heartbeat且freq字段携带发送周期。七、心跳在监控生态中的角色与 FAQ7.1 监控系统如何利用心跳Flower、celery events、celery events --camera等监控工具通过监听事件流中的worker-online/worker-heartbeat/worker-offline事件来构建 Worker 在线状态视图。判断逻辑通常为收到worker-online→ Worker 标记为在线持续收到worker-heartbeat→ Worker 存活且可从事件字段读取active活跃请求数、processed处理任务数、loadavg系统负载用于负载展示超过若干个freq周期未收到心跳 → Worker 被判定为离线或失联。因此--heartbeat-interval的大小直接决定监控系统的故障发现延迟与事件流量之间的权衡。7.2 常见问题Q1--without-heartbeat与send_task_events是什么关系二者独立。--without-heartbeat只关闭心跳 bootstepHeart.enabled Falsesend_task_events旧名celery_send_events控制是否发送任务级事件由Eventsbootstep 的groups过滤实现。关闭心跳不会自动关闭任务事件反之亦然。Q2为什么建议不要在--without-heartbeat下依赖监控心跳是监控系统感知 Worker 存活状态的主要信号源在 gossip 关闭时几乎是唯一来源。Events.__init__中send_events的计算保证了 heartbeat 关闭时事件分发器仍可用但不会再产生周期性心跳监控端将因缺少心跳事件而无法及时感知 Worker 掉线。Q3心跳事件发送失败会怎样_send默认retryTrueworker-offline除外发送失败会由事件分发器按既有重试策略处理worker-offline使用retryFalse避免在连接已不可用时做无谓重试。Q4事件心跳能用于判断任务正在运行吗可以间接判断。worker-heartbeat事件携带active当前活跃请求数字段监控端可据此感知 Worker 是否正在处理任务但它不是任务粒度的事件——任务粒度应依赖task-started/task-succeeded等任务事件需开启send_task_events。八、小结从配置到源码的完整链路层级文件职责CLI 选项celery/bin/worker.py--without-heartbeat、--heartbeat-intervalBootstepcelery/worker/consumer/heart.pyHeartenabled 开关、依赖Events、生命周期委托服务实现celery/worker/heartbeat.py周期发送worker-heartbeat附带active/processed/loadavg事件分发celery/worker/consumer/events.py创建EventDispatcher联动 AMQP 心跳配置默认值celery/app/defaults.pybroker_heartbeat120、broker_heartbeat_checkrate3.0AMQP 级信号钩子celery/signals.pyheartbeat_sent有接收者时才分发测试印证t/unit/worker/test_consumer.py、t/unit/worker/test_heartbeat.py锁定 bootstep 契约与事件序列理解这条链路后你可以精准地控制 Worker 心跳行为用--heartbeat-interval调节故障发现灵敏度用--without-heartbeat在不需要监控时降低事件开销通过heartbeat_sent信号在本进程内挂钩子或直接消费worker-heartbeat事件构建自定义的 Worker 监控大盘。【免费下载链接】celeryDistributed Task Queue (development branch)项目地址: https://gitcode.com/gh_mirrors/ce/celery创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考