任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载导读本文以 Celery 仓库中的 docs/reference/celery.beat.rst API 参考文档为骨架深入剖析其核心对象celery.beat模块源码位于 celery/beat.py。该模块是celery beat周期任务调度器的实现主体负责按beat_schedule配置在固定间隔内将任务发送到 Broker。读完本文你将掌握ScheduleEntry、Scheduler、PersistentScheduler、Service、EmbeddedService等核心类的职责与协作方式理解调度堆heap与 tick 循环的工作原理、持久化与同步机制以及相关配置项、CLI 命令与信号的完整用法。一、模块定位celery.beat 是什么celery.beat是 Celery 的周期任务调度器periodic task scheduler模块。正如其模块 docstring 所言——The periodic task scheduler——它独立于 worker 运行按预定义的时间表周期性地把任务发送到消息队列由 worker 消费执行。在 docs/userguide/periodic-tasks.rst 中:program:celery beat 被描述为a scheduler; It kicks off tasks at regular intervals。默认情况下调度条目entries取自beat_schedule配置项。而 celery/apps/beat.py 则明确写道它是celery.beat模块的program-version——即负责把celery.beat真正跑成一个应用程序包括安装信号处理器、创建 pidfile、设置进程标题、配置日志等。celery.beat模块导出的公开 API见 celery/beat.py 的__all__为SchedulingError调度过程中抛出的异常ScheduleEntry调度表中的一个条目Scheduler周期任务调度器基类PersistentScheduler基于shelve数据库持久化的调度器ServiceCelery 周期任务服务EmbeddedService嵌入式时钟服务工厂函数另有未列入__all__的辅助类BeatLazyFunc惰性函数包装与命名元组event_t namedtuple(event_t, (time, priority, entry))后者是调度堆中每个元素的载体。二、ScheduleEntry调度表的基本单元ScheduleEntry源码见 celery/beat.py代表调度表中的一条记录即某个任务按照某个时间表周期性执行的完整描述。2.1 字段定义字段类型说明namestr调度条目的名称唯一标识taskstr要调度的任务名schedule~celery.schedules.schedule时间表对象如crontab、timedelta等经maybe_schedule统一转换argsTuple传给任务的位置参数kwargsDict传给任务的关键字参数optionsDict任务执行选项last_run_atdatetime上次计划执行的时间total_run_countint该条目已被调度的总次数relativebool时间是否相对于服务器启动时刻构造函数签名celery/beat.pydef __init__(self, nameNone, taskNone, last_run_atNone, total_run_countNone, scheduleNone, args(), kwargsNone, optionsNone, relativeFalse, appNone):注意几个实现细节schedule通过maybe_schedule(schedule, relative, appself.app)转换因此既可以直接传~celery.schedules.schedule实例也可以传整数秒数或datetime.timedelta详见 celery/schedules.py 的schedule类。last_run_at缺省时取self.schedule.now()若 schedule 存在否则取self.app.now()。该类的__iter__返回vars(self).items()支持dict(self)形式的序列化重建_next_instance正是这样构造下一个实例的。2.2 核心行为_next_instance()/next()/__next__返回一个新实例仅更新last_run_at默认取当前时间并将total_run_count加 1。调度器每次真正派发一个条目后就用它替换旧条目从而推进调度状态。is_due()委托给self.schedule.is_due(self.last_run_at)返回(is_due, next_time_to_run)二元组——是否到期以及距离下次运行的时间秒。update(other)只更新可编辑字段task、schedule、args、kwargs、options用于磁盘上的旧条目与配置中新增定义合并。__eq__/editable_fields_equal相等性只比较上述可编辑字段而__lt__则退回id(self) id(other)正如注释所解释的——在调度堆中排序由元组(time, priority, entry)的前两个成员决定堆顶竞争最后才轮到比较 entry因此此时顺序随机即可。从源码结构看ScheduleEntry的可编辑字段与运行状态字段last_run_at / total_run_count的分离是刻意设计前者用于合并外部配置更新后者用于跟踪调度进度。对应的单元测试位于 t/unit/app/test_beat.py其中test_next、test_is_due、test_update、test_repr、test_reduce、test_lt等用例直接覆盖了上述行为。三、BeatLazyFunc调度参数中的惰性求值BeatLazyFunccelery/beat.py用于在beat_schedule中声明发送任务前才调用的惰性函数。其典型场景是某些参数如当前时间必须在任务真正派发那一刻才求值而不能在配置加载时固定下来。官方 docstring 给出的示例beat_schedule { test-every-5-minutes: { task: test, schedule: 300, kwargs: { current: BeatCallBack(datetime.datetime.now) } } }实现要点构造时接收func及预置的args/kwargs保存在self._func_params。定义__call__与delay()二者都会在调用时执行self._func(*args, **kwargs)返回真实值。它真正被消费的位置在Scheduler.apply_async内部见下文 4.4 节_evaluate_entry_args与_evaluate_entry_kwargscelery/beat.py会遍历条目的args/kwargs把其中所有BeatLazyFunc实例替换为其调用结果。t/unit/app/test_beat.py的test_beat_lazy_func对该机制进行了验证。四、Scheduler调度器核心Schedulercelery/beat.py是所有调度器的基类实现了完整的调度主循环、到期判定、任务派发与同步机制。celery beat程序可能出于内省目的多次实例化本类此时会传入lazyTrue——因此文档特别强调子类在lazy参数下必须保持幂等。4.1 关键类属性属性默认值说明EntryScheduleEntry使用的条目类型子类可覆写max_intervalDEFAULT_MAX_INTERVAL 3005 分钟两次检查调度表之间的最大睡眠秒数sync_every3 * 603 分钟多久强制同步一次调度表sync_every_tasksNone每派发多少任务强制同步一次默认不启用4.2 初始化与配置解析构造函数celery/beat.pydef __init__(self, app, scheduleNone, max_intervalNone, ProducerNone, lazyFalse, sync_every_tasksNone, **kwargs): self.app app self.data maybe_evaluate({} if schedule is None else schedule) self.max_interval (max_interval or app.conf.beat_max_loop_interval or self.max_interval) self.Producer Producer or app.amqp.Producer ... self.sync_every_tasks ( app.conf.beat_sync_every if sync_every_tasks is None else sync_every_tasks) if not lazy: self.setup_schedule()注意配置优先级的实际体现max_interval依次取构造函数参数 →app.conf.beat_max_loop_interval→ 类属性默认值 300 秒sync_every_tasks缺省时取app.conf.beat_sync_every。这直接对应 celery/app/defaults.py 中beat命名空间的配置定义。setup_schedule()celery/beat.py做两件事install_default_entries(self.data)若result_expires已配置且后端不支持自动过期则自动注册celery.backend_cleanup任务默认crontab(0, 4, *)即每天凌晨 4 点options带expires: 12 * 3600用于清理过期结果。merge_inplace(self.app.conf.beat_schedule)将beat_schedule配置合并进调度表。merge_inplacecelery/beat.py的合并语义值得注意计算配置键集合B与现有调度键集合A的对称差A ^ B从调度表中移除磁盘上存在但配置里已删除的条目对B中的每个键若已存在则调用schedule[key].update(entry)仅刷新可编辑字段否则新建条目。这样既能应用配置变更又不丢失last_run_at/total_run_count等运行状态。4.3 调度堆heap与 tick 循环Scheduler的核心数据结构是一个最小堆self._heap堆元素为event_t(time, priority, entry)三元组按触发时间戳排序。populate_heap()celery/beat.py遍历self.schedule.values()对每个条目计算is_due, next_call_delay entry.is_due()若已到期则时间戳按 0 计算否则按next_call_delay计算随后heapify。时间戳由_when()生成将last_run_at归一化为 UTC 后加上微秒与经过adjust()默认减去 0.010 秒漂移补偿的延迟值。tick()celery/beat.py——每次调用执行一个到期任务是调度的主循环体返回下次调用前应等待的秒数建议。其流程如下若堆为空或调度表发生变化schedules_equal比较键集合与各条目可编辑字段则重建堆。取堆顶事件若event[0] now说明最早的到期时间在未来返回min(event[0] - now, max_interval)作为睡眠建议。若堆顶已到期heappop弹出并校验身份后reserve(entry)即self.schedule[entry.name] next(entry)推进运行状态并apply_entry派发任务然后把next_entry按新的触发时间重新入堆返回 0立即继续下一轮。若堆顶已到期但条目自身要求稍后重试将其按重试时间重新入堆reschedule_delay避免该条目长期占据堆顶、饿死后续条目——这一点在代码注释中明确引用了https://github.com/celery/celery/issues/7649的问题背景。tick的这些分支在 t/unit/app/test_beat.py 中有大量测试覆盖例如test_due_tick、test_due_tick_returns_delay_when_heap_top_changed、test_pending_tick、test_honors_max_interval、test_not_due_top_entry_is_rescheduled_behind_due_entry以及针对错过 cron 截止期限的test_tick_dispatches_missed_cron_within_deadline_non_uniform等。4.4 任务派发apply_entry 与 apply_asyncapply_entrycelery/beat.py负责实际发送到期任务记录 info 级日志Scheduler: Sending due task %s (%s)派发失败时记录 error 日志并继续不会让整个 beat 崩溃。真正的工作在apply_asynccelery/beat.py中完成要点如下先reserve推进时间戳与计数注释强调要在真正执行前完成避免异常导致永远重复调度。对entry.args/entry.kwargs执行惰性求值见第三节BeatLazyFunc。给消息注入自定义 headeroptions.setdefault(headers, {})[celery_beat_task] True用于标识消息来源于 Celery Beat。若任务已注册走task.apply_async(...)否则退回self.send_task(...)即app.send_task。派发后递增_tasks_since_sync若should_sync()为真则执行_do_sync()同步调度表。任何异常都会被包装为SchedulingError重新抛出消息形如Couldnt apply scheduled task {name}: {exc}。apply_async的行为在测试中有直接印证test_apply_async_uses_registered_task_instances、test_apply_async_with_null_args、test_apply_async_with_null_args_set_to_none等用例t/unit/app/test_beat.py。4.5 同步机制should_sync / sync / closeshould_sync()celery/beat.py满足任一条件即需同步——距上次同步超过sync_every秒或启用了sync_every_tasks时自上次同步以来派发任务数达到阈值。_do_sync()调用self.sync()并重置_last_sync与_tasks_since_sync。基类的sync()为空实现内存型调度器无需落盘close()也仅调用sync()PersistentScheduler会覆写二者。4.6 连接与 Producerconnection与producer均为cached_propertycelery/beat.pycached_property def connection(self): return self.app.connection_for_write() cached_property def producer(self): return self.Producer(self._ensure_connected(), auto_declareFalse)_ensure_connected()通过connection.ensure_connection(_error_handler, self.app.conf.broker_connection_max_retries)建立连接连接失败时以beat: Connection error: %s. Trying again in %s seconds...记录错误并重试重试上限由broker_connection_max_retries控制。五、PersistentScheduler基于 shelve 的持久化调度器PersistentSchedulercelery/beat.py是默认使用的调度器配置beat_scheduler的默认值即为celery.beat:PersistentScheduler见 celery/app/defaults.py它把调度表持久化到shelve数据库中默认文件名celerybeat-schedule。5.1 文件与损坏恢复persistence shelveknown_suffixes (, .db, .dat, .bak, .dir)——_remove_db()会按这些后缀逐一尝试删除用platforms.ignore_errno(errno.ENOENT)忽略文件不存在。_open_schedule()以writebackTrue打开数据库。setup_schedule()打开数据库并读取keys()以触发潜在损坏错误若失败则记录Removing corrupted schedule file %r: %r并删除数据库重建——代码注释特别提到 bsddb 的DBPageNotFoundError这类打开成功但首次取键才报错的损坏场景。5.2 时区 / UTC 变更重置持久化数据库中会额外保存__version__、tz、utc_enabled三个元数据字段若存储的tz与当前app.conf.timezone不同或存储的utc_enabled与app.conf.enable_utc不同则警告Reset: Timezone changed from %r to %r/Reset: UTC changed from %s to %s并clear()整个数据库——因为时间表语义依赖时区变更时必须重置以免按错误时区触发。_create_schedule()中的升级逻辑对旧版本数据库做了兼容缺少__version__字段2.2.2 之前→ 重置缺少tz3.0.8 之前→ 重置缺少utc_enabled3.0.9 之前→ 重置。5.3 覆盖的接口schedule属性改为self._store[entries]的读写。sync()调用self._store.sync()close()在sync()后关闭self._store。info返回f . db - {self.schedule_filename}用于启动横幅展示。六、Service 与 EmbeddedService把调度器跑起来6.1 ServiceServicecelery/beat.py把调度器封装为可启动/停止的服务是celery beat命令行与嵌入模式共用的运行时外壳scheduler_cls PersistentScheduler默认构造函数中的schedule_filename缺省取app.conf.beat_schedule_filenamemax_interval缺省取app.conf.beat_max_loop_interval。使用两个threading.Event_is_shutdown与_is_stopped协调启停。start(embedded_processFalse)的主循环while not self._is_shutdown.is_set(): interval self.scheduler.tick() if interval and interval 0.0: debug(beat: Waking up %s., humanize_seconds(interval, prefixin )) time.sleep(interval) if self.scheduler.should_sync(): self.scheduler._do_sync()启动时发送beat_init信号若为嵌入进程还发送beat_embedded_init信号并将进程标题设为celery beat捕获KeyboardInterrupt/SystemExit后置_is_shutdown最终self.sync()关闭调度器并置_is_stopped。stop(waitFalse)置_is_shutdown可选阻塞等待_is_stopped。get_scheduler(lazyFalse, extension_namespacecelery.beat_schedulers)通过load_extension_class_names加载celery.beat_schedulers命名空间下的第三方调度器别名再用symbol_by_name解析scheduler_cls——这是自定义调度器如数据库调度器被beat_scheduler配置引用的解析入口。6.2 EmbeddedServiceEmbeddedServicecelery/beat.py是返回嵌入式时钟服务的工厂函数def EmbeddedService(app, max_intervalNone, **kwargs): if kwargs.pop(thread, False) or _Process is None: # Need short max interval to be able to stop thread # in reasonable time. return _Threaded(app, max_interval1, **kwargs) return _Process(app, max_intervalmax_interval, **kwargs)默认使用多进程_Process基于billiard的Processrun()中会reset_signals、关闭标准输入输出与日志文件描述符并调用service.start(embedded_processTrue)。传threadTrue或平台不支持多进程时回退到_Threaded其max_interval被强制设为 1 秒以便线程能在合理时间内被停止线程名为Beat且为 daemon 线程。这解释了嵌入式模式如 Django 中通过app.Beat或直接调用EmbeddedService在应用进程内运行 beat的两种形态也对应 celery/signals.py 中beat_init与beat_embedded_init两个信号的存在意义。七、配置项速查beat 命名空间以下配置定义于 celery/app/defaults.py 的beat命名空间配置键默认值类型作用beat_max_loop_interval0float调度循环两次检查之间的最大睡眠秒数0 表示使用调度器类默认值300 秒beat_schedule{}dict周期任务调度表键为条目名值为{task, schedule, args, kwargs, options}字典beat_schedulercelery.beat:PersistentSchedulerstr调度器类可为点路径或注册在celery.beat_schedulers命名空间的别名beat_schedule_filenamecelerybeat-schedulestr调度数据库文件名beat_sync_every0int每派发 N 个任务强制同步调度表0 表示关闭仅按时间同步beat_cron_starting_deadlineNoneint错过 cron 触发后的补偿截止期限秒相关分支见tick()与测试用例test_tick_dispatches_missed_cron_within_deadline_non_uniform另外Scheduler中sync_every 3 * 60的按时间同步默认间隔是类属性而非配置项beat_schedule的完整写法可参考 docs/userguide/periodic-tasks.rst 中app.conf.beat_schedule的示例支持crontab与整数秒/timedelta两种 schedule 类型。八、命令行celery beat8.1 启动命令celery -A proj beat默认从当前目录读取celerybeat-schedule数据库文件若不存在则创建。celery.beat模块在此处的角色是被驱动方——CLI 层在 celery/bin/beat.py 中定义应用层封装在 celery/apps/beat.py 的Beat类最终调用beat.Service即celery.beat的Service。8.2 常用选项Beat Options见 celery/bin/beat.py选项说明--detach后台守护进程方式运行-s, --schedule调度数据库路径默认取beat_schedule_filename配置默认celerybeat-schedule扩展名.db会被追加-S, --scheduler使用的调度器类默认取beat_scheduler配置即celery.beat:PersistentScheduler--max-interval调度迭代之间最大睡眠秒数int对应max_interval-l, --loglevel日志级别默认WARNING-f, --logfile、--pidfile、--uid、--gid、--umask、--workdir继承自守护命令的通用选项此外还支持-C无彩色输出与额外配置参数透传ctx.args中未被识别的内容会交给app.config_from_cmdline(ctx.args)解析解析失败时抛出click.UsageError。8.3 启动流程与信号处理Beat.run()celery/apps/beat.py依次执行打印celery beat v{VERSION_BANNER} is starting.横幅init_loader()app.loader.init_worker()导入任务模块并app.finalize()设置进程标题celery beatstart_scheduler()若指定pidfile则platforms.create_pidlock构造Service并打印启动信息STARTUP_INFO_FMT展示 broker、loader、scheduler、logfile、maxinterval 等设置日志可选设置全局 socket 超时默认 30 秒安装同步信号处理器后service.start()。install_sync_handlercelery/apps/beat.py为SIGTERM与SIGINT安装处理器收到信号时先service.sync()保存调度表再抛SystemExit——确保关闭时调度状态如last_run_at、total_run_count不丢失。九、信号钩子celery.beat相关信号定义于 celery/signals.pybeat_init在Service.start()中、主循环开始前发送signals.beat_init.send(senderself)可用于初始化资源。beat_embedded_init仅在嵌入式进程模式embedded_processTrue下发送用于区分独立进程与嵌入应用进程两种运行形态。十、扩展调度器与测试验证10.1 自定义调度器Service.get_scheduler的extension_namespacecelery.beat_schedulers表明第三方调度器例如将调度表存入数据库的django_celery_beat.schedulers:DatabaseScheduler通过注册到该命名空间即可用配置或-S选项引用。自定义调度器通常继承Scheduler覆写setup_schedule、sync、close与schedule属性即可参考PersistentScheduler的模式。10.2 测试支撑模块级行为在 t/unit/app/test_beat.py 中有完整覆盖可作为理解实现语义的活文档代表性用例包括惰性参数求值test_beat_lazy_func条目推进test_next、test_reducetick 分支test_due_tick、test_pending_tick、test_honors_max_interval、test_ticks_schedule_change、test_not_due_top_entry_is_rescheduled_behind_due_entry同步计数test_should_sync、test_sync_task_counter_resets_on_do_sync默认条目与合并test_install_default_entries、test_merge_inplace错过 cron 的截止期限补偿test_tick_dispatches_missed_cron_within_deadline_non_uniform等。结语celery.beat模块以条目ScheduleEntry 调度器Scheduler 服务Service三层结构将周期任务调度抽象为清晰可扩展的组件堆驱动的 tick 循环保证调度效率持久化与同步机制保证状态不丢失beat_schedule配置与celery beat命令提供开箱即用的使用入口而beat_init/beat_embedded_init信号与celery.beat_schedulers扩展命名空间则为其在框架内的定制留足空间。理解这一模块是深入使用与扩展 Celery 周期任务能力的关键一步。赞分享任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载相关推荐Celery 周期性任务Periodic Tasks完整指南beat 调度器、Crontab 与 Solar 调度实战Celery 周期性任务Periodic Tasks完整指南beat 调度器、Crontab 与 Solar 调度实战 导读 本文以 Celery 官方用任务调度后端消息队列雀魂数据分析终极指南如何用免费工具快速提升麻将水平雀魂数据分析终极指南如何用免费工具快速提升麻将水平 想要从雀魂麻将新手成长为高手吗雀魂牌谱屋amae koromo就是你需要的秘密武器这款完全免费的开CANNAscend人工智能任务调度celery beat 命令完全指南配置、调度器与源码级原理剖析celery beat 命令完全指南配置、调度器与源码级原理剖析 导读 celery beat 是 Celery 分布式任务队列内置的周期任务调度器负责任务调度后端消息队列上一篇YYText手势处理终极指南双击放大与长按操作详解下一篇DictaLM 2.0性能评测希伯来语各项NLP任务的表现分析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Celery Beat 周期任务调度器：celery.beat 模块架构与源码级解析
任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载导读本文以 Celery 仓库中的 docs/reference/celery.beat.rst API 参考文档为骨架深入剖析其核心对象celery.beat模块源码位于 celery/beat.py。该模块是celery beat周期任务调度器的实现主体负责按beat_schedule配置在固定间隔内将任务发送到 Broker。读完本文你将掌握ScheduleEntry、Scheduler、PersistentScheduler、Service、EmbeddedService等核心类的职责与协作方式理解调度堆heap与 tick 循环的工作原理、持久化与同步机制以及相关配置项、CLI 命令与信号的完整用法。一、模块定位celery.beat 是什么celery.beat是 Celery 的周期任务调度器periodic task scheduler模块。正如其模块 docstring 所言——The periodic task scheduler——它独立于 worker 运行按预定义的时间表周期性地把任务发送到消息队列由 worker 消费执行。在 docs/userguide/periodic-tasks.rst 中:program:celery beat 被描述为a scheduler; It kicks off tasks at regular intervals。默认情况下调度条目entries取自beat_schedule配置项。而 celery/apps/beat.py 则明确写道它是celery.beat模块的program-version——即负责把celery.beat真正跑成一个应用程序包括安装信号处理器、创建 pidfile、设置进程标题、配置日志等。celery.beat模块导出的公开 API见 celery/beat.py 的__all__为SchedulingError调度过程中抛出的异常ScheduleEntry调度表中的一个条目Scheduler周期任务调度器基类PersistentScheduler基于shelve数据库持久化的调度器ServiceCelery 周期任务服务EmbeddedService嵌入式时钟服务工厂函数另有未列入__all__的辅助类BeatLazyFunc惰性函数包装与命名元组event_t namedtuple(event_t, (time, priority, entry))后者是调度堆中每个元素的载体。二、ScheduleEntry调度表的基本单元ScheduleEntry源码见 celery/beat.py代表调度表中的一条记录即某个任务按照某个时间表周期性执行的完整描述。2.1 字段定义字段类型说明namestr调度条目的名称唯一标识taskstr要调度的任务名schedule~celery.schedules.schedule时间表对象如crontab、timedelta等经maybe_schedule统一转换argsTuple传给任务的位置参数kwargsDict传给任务的关键字参数optionsDict任务执行选项last_run_atdatetime上次计划执行的时间total_run_countint该条目已被调度的总次数relativebool时间是否相对于服务器启动时刻构造函数签名celery/beat.pydef __init__(self, nameNone, taskNone, last_run_atNone, total_run_countNone, scheduleNone, args(), kwargsNone, optionsNone, relativeFalse, appNone):注意几个实现细节schedule通过maybe_schedule(schedule, relative, appself.app)转换因此既可以直接传~celery.schedules.schedule实例也可以传整数秒数或datetime.timedelta详见 celery/schedules.py 的schedule类。last_run_at缺省时取self.schedule.now()若 schedule 存在否则取self.app.now()。该类的__iter__返回vars(self).items()支持dict(self)形式的序列化重建_next_instance正是这样构造下一个实例的。2.2 核心行为_next_instance()/next()/__next__返回一个新实例仅更新last_run_at默认取当前时间并将total_run_count加 1。调度器每次真正派发一个条目后就用它替换旧条目从而推进调度状态。is_due()委托给self.schedule.is_due(self.last_run_at)返回(is_due, next_time_to_run)二元组——是否到期以及距离下次运行的时间秒。update(other)只更新可编辑字段task、schedule、args、kwargs、options用于磁盘上的旧条目与配置中新增定义合并。__eq__/editable_fields_equal相等性只比较上述可编辑字段而__lt__则退回id(self) id(other)正如注释所解释的——在调度堆中排序由元组(time, priority, entry)的前两个成员决定堆顶竞争最后才轮到比较 entry因此此时顺序随机即可。从源码结构看ScheduleEntry的可编辑字段与运行状态字段last_run_at / total_run_count的分离是刻意设计前者用于合并外部配置更新后者用于跟踪调度进度。对应的单元测试位于 t/unit/app/test_beat.py其中test_next、test_is_due、test_update、test_repr、test_reduce、test_lt等用例直接覆盖了上述行为。三、BeatLazyFunc调度参数中的惰性求值BeatLazyFunccelery/beat.py用于在beat_schedule中声明发送任务前才调用的惰性函数。其典型场景是某些参数如当前时间必须在任务真正派发那一刻才求值而不能在配置加载时固定下来。官方 docstring 给出的示例beat_schedule { test-every-5-minutes: { task: test, schedule: 300, kwargs: { current: BeatCallBack(datetime.datetime.now) } } }实现要点构造时接收func及预置的args/kwargs保存在self._func_params。定义__call__与delay()二者都会在调用时执行self._func(*args, **kwargs)返回真实值。它真正被消费的位置在Scheduler.apply_async内部见下文 4.4 节_evaluate_entry_args与_evaluate_entry_kwargscelery/beat.py会遍历条目的args/kwargs把其中所有BeatLazyFunc实例替换为其调用结果。t/unit/app/test_beat.py的test_beat_lazy_func对该机制进行了验证。四、Scheduler调度器核心Schedulercelery/beat.py是所有调度器的基类实现了完整的调度主循环、到期判定、任务派发与同步机制。celery beat程序可能出于内省目的多次实例化本类此时会传入lazyTrue——因此文档特别强调子类在lazy参数下必须保持幂等。4.1 关键类属性属性默认值说明EntryScheduleEntry使用的条目类型子类可覆写max_intervalDEFAULT_MAX_INTERVAL 3005 分钟两次检查调度表之间的最大睡眠秒数sync_every3 * 603 分钟多久强制同步一次调度表sync_every_tasksNone每派发多少任务强制同步一次默认不启用4.2 初始化与配置解析构造函数celery/beat.pydef __init__(self, app, scheduleNone, max_intervalNone, ProducerNone, lazyFalse, sync_every_tasksNone, **kwargs): self.app app self.data maybe_evaluate({} if schedule is None else schedule) self.max_interval (max_interval or app.conf.beat_max_loop_interval or self.max_interval) self.Producer Producer or app.amqp.Producer ... self.sync_every_tasks ( app.conf.beat_sync_every if sync_every_tasks is None else sync_every_tasks) if not lazy: self.setup_schedule()注意配置优先级的实际体现max_interval依次取构造函数参数 →app.conf.beat_max_loop_interval→ 类属性默认值 300 秒sync_every_tasks缺省时取app.conf.beat_sync_every。这直接对应 celery/app/defaults.py 中beat命名空间的配置定义。setup_schedule()celery/beat.py做两件事install_default_entries(self.data)若result_expires已配置且后端不支持自动过期则自动注册celery.backend_cleanup任务默认crontab(0, 4, *)即每天凌晨 4 点options带expires: 12 * 3600用于清理过期结果。merge_inplace(self.app.conf.beat_schedule)将beat_schedule配置合并进调度表。merge_inplacecelery/beat.py的合并语义值得注意计算配置键集合B与现有调度键集合A的对称差A ^ B从调度表中移除磁盘上存在但配置里已删除的条目对B中的每个键若已存在则调用schedule[key].update(entry)仅刷新可编辑字段否则新建条目。这样既能应用配置变更又不丢失last_run_at/total_run_count等运行状态。4.3 调度堆heap与 tick 循环Scheduler的核心数据结构是一个最小堆self._heap堆元素为event_t(time, priority, entry)三元组按触发时间戳排序。populate_heap()celery/beat.py遍历self.schedule.values()对每个条目计算is_due, next_call_delay entry.is_due()若已到期则时间戳按 0 计算否则按next_call_delay计算随后heapify。时间戳由_when()生成将last_run_at归一化为 UTC 后加上微秒与经过adjust()默认减去 0.010 秒漂移补偿的延迟值。tick()celery/beat.py——每次调用执行一个到期任务是调度的主循环体返回下次调用前应等待的秒数建议。其流程如下若堆为空或调度表发生变化schedules_equal比较键集合与各条目可编辑字段则重建堆。取堆顶事件若event[0] now说明最早的到期时间在未来返回min(event[0] - now, max_interval)作为睡眠建议。若堆顶已到期heappop弹出并校验身份后reserve(entry)即self.schedule[entry.name] next(entry)推进运行状态并apply_entry派发任务然后把next_entry按新的触发时间重新入堆返回 0立即继续下一轮。若堆顶已到期但条目自身要求稍后重试将其按重试时间重新入堆reschedule_delay避免该条目长期占据堆顶、饿死后续条目——这一点在代码注释中明确引用了https://github.com/celery/celery/issues/7649的问题背景。tick的这些分支在 t/unit/app/test_beat.py 中有大量测试覆盖例如test_due_tick、test_due_tick_returns_delay_when_heap_top_changed、test_pending_tick、test_honors_max_interval、test_not_due_top_entry_is_rescheduled_behind_due_entry以及针对错过 cron 截止期限的test_tick_dispatches_missed_cron_within_deadline_non_uniform等。4.4 任务派发apply_entry 与 apply_asyncapply_entrycelery/beat.py负责实际发送到期任务记录 info 级日志Scheduler: Sending due task %s (%s)派发失败时记录 error 日志并继续不会让整个 beat 崩溃。真正的工作在apply_asynccelery/beat.py中完成要点如下先reserve推进时间戳与计数注释强调要在真正执行前完成避免异常导致永远重复调度。对entry.args/entry.kwargs执行惰性求值见第三节BeatLazyFunc。给消息注入自定义 headeroptions.setdefault(headers, {})[celery_beat_task] True用于标识消息来源于 Celery Beat。若任务已注册走task.apply_async(...)否则退回self.send_task(...)即app.send_task。派发后递增_tasks_since_sync若should_sync()为真则执行_do_sync()同步调度表。任何异常都会被包装为SchedulingError重新抛出消息形如Couldnt apply scheduled task {name}: {exc}。apply_async的行为在测试中有直接印证test_apply_async_uses_registered_task_instances、test_apply_async_with_null_args、test_apply_async_with_null_args_set_to_none等用例t/unit/app/test_beat.py。4.5 同步机制should_sync / sync / closeshould_sync()celery/beat.py满足任一条件即需同步——距上次同步超过sync_every秒或启用了sync_every_tasks时自上次同步以来派发任务数达到阈值。_do_sync()调用self.sync()并重置_last_sync与_tasks_since_sync。基类的sync()为空实现内存型调度器无需落盘close()也仅调用sync()PersistentScheduler会覆写二者。4.6 连接与 Producerconnection与producer均为cached_propertycelery/beat.pycached_property def connection(self): return self.app.connection_for_write() cached_property def producer(self): return self.Producer(self._ensure_connected(), auto_declareFalse)_ensure_connected()通过connection.ensure_connection(_error_handler, self.app.conf.broker_connection_max_retries)建立连接连接失败时以beat: Connection error: %s. Trying again in %s seconds...记录错误并重试重试上限由broker_connection_max_retries控制。五、PersistentScheduler基于 shelve 的持久化调度器PersistentSchedulercelery/beat.py是默认使用的调度器配置beat_scheduler的默认值即为celery.beat:PersistentScheduler见 celery/app/defaults.py它把调度表持久化到shelve数据库中默认文件名celerybeat-schedule。5.1 文件与损坏恢复persistence shelveknown_suffixes (, .db, .dat, .bak, .dir)——_remove_db()会按这些后缀逐一尝试删除用platforms.ignore_errno(errno.ENOENT)忽略文件不存在。_open_schedule()以writebackTrue打开数据库。setup_schedule()打开数据库并读取keys()以触发潜在损坏错误若失败则记录Removing corrupted schedule file %r: %r并删除数据库重建——代码注释特别提到 bsddb 的DBPageNotFoundError这类打开成功但首次取键才报错的损坏场景。5.2 时区 / UTC 变更重置持久化数据库中会额外保存__version__、tz、utc_enabled三个元数据字段若存储的tz与当前app.conf.timezone不同或存储的utc_enabled与app.conf.enable_utc不同则警告Reset: Timezone changed from %r to %r/Reset: UTC changed from %s to %s并clear()整个数据库——因为时间表语义依赖时区变更时必须重置以免按错误时区触发。_create_schedule()中的升级逻辑对旧版本数据库做了兼容缺少__version__字段2.2.2 之前→ 重置缺少tz3.0.8 之前→ 重置缺少utc_enabled3.0.9 之前→ 重置。5.3 覆盖的接口schedule属性改为self._store[entries]的读写。sync()调用self._store.sync()close()在sync()后关闭self._store。info返回f . db - {self.schedule_filename}用于启动横幅展示。六、Service 与 EmbeddedService把调度器跑起来6.1 ServiceServicecelery/beat.py把调度器封装为可启动/停止的服务是celery beat命令行与嵌入模式共用的运行时外壳scheduler_cls PersistentScheduler默认构造函数中的schedule_filename缺省取app.conf.beat_schedule_filenamemax_interval缺省取app.conf.beat_max_loop_interval。使用两个threading.Event_is_shutdown与_is_stopped协调启停。start(embedded_processFalse)的主循环while not self._is_shutdown.is_set(): interval self.scheduler.tick() if interval and interval 0.0: debug(beat: Waking up %s., humanize_seconds(interval, prefixin )) time.sleep(interval) if self.scheduler.should_sync(): self.scheduler._do_sync()启动时发送beat_init信号若为嵌入进程还发送beat_embedded_init信号并将进程标题设为celery beat捕获KeyboardInterrupt/SystemExit后置_is_shutdown最终self.sync()关闭调度器并置_is_stopped。stop(waitFalse)置_is_shutdown可选阻塞等待_is_stopped。get_scheduler(lazyFalse, extension_namespacecelery.beat_schedulers)通过load_extension_class_names加载celery.beat_schedulers命名空间下的第三方调度器别名再用symbol_by_name解析scheduler_cls——这是自定义调度器如数据库调度器被beat_scheduler配置引用的解析入口。6.2 EmbeddedServiceEmbeddedServicecelery/beat.py是返回嵌入式时钟服务的工厂函数def EmbeddedService(app, max_intervalNone, **kwargs): if kwargs.pop(thread, False) or _Process is None: # Need short max interval to be able to stop thread # in reasonable time. return _Threaded(app, max_interval1, **kwargs) return _Process(app, max_intervalmax_interval, **kwargs)默认使用多进程_Process基于billiard的Processrun()中会reset_signals、关闭标准输入输出与日志文件描述符并调用service.start(embedded_processTrue)。传threadTrue或平台不支持多进程时回退到_Threaded其max_interval被强制设为 1 秒以便线程能在合理时间内被停止线程名为Beat且为 daemon 线程。这解释了嵌入式模式如 Django 中通过app.Beat或直接调用EmbeddedService在应用进程内运行 beat的两种形态也对应 celery/signals.py 中beat_init与beat_embedded_init两个信号的存在意义。七、配置项速查beat 命名空间以下配置定义于 celery/app/defaults.py 的beat命名空间配置键默认值类型作用beat_max_loop_interval0float调度循环两次检查之间的最大睡眠秒数0 表示使用调度器类默认值300 秒beat_schedule{}dict周期任务调度表键为条目名值为{task, schedule, args, kwargs, options}字典beat_schedulercelery.beat:PersistentSchedulerstr调度器类可为点路径或注册在celery.beat_schedulers命名空间的别名beat_schedule_filenamecelerybeat-schedulestr调度数据库文件名beat_sync_every0int每派发 N 个任务强制同步调度表0 表示关闭仅按时间同步beat_cron_starting_deadlineNoneint错过 cron 触发后的补偿截止期限秒相关分支见tick()与测试用例test_tick_dispatches_missed_cron_within_deadline_non_uniform另外Scheduler中sync_every 3 * 60的按时间同步默认间隔是类属性而非配置项beat_schedule的完整写法可参考 docs/userguide/periodic-tasks.rst 中app.conf.beat_schedule的示例支持crontab与整数秒/timedelta两种 schedule 类型。八、命令行celery beat8.1 启动命令celery -A proj beat默认从当前目录读取celerybeat-schedule数据库文件若不存在则创建。celery.beat模块在此处的角色是被驱动方——CLI 层在 celery/bin/beat.py 中定义应用层封装在 celery/apps/beat.py 的Beat类最终调用beat.Service即celery.beat的Service。8.2 常用选项Beat Options见 celery/bin/beat.py选项说明--detach后台守护进程方式运行-s, --schedule调度数据库路径默认取beat_schedule_filename配置默认celerybeat-schedule扩展名.db会被追加-S, --scheduler使用的调度器类默认取beat_scheduler配置即celery.beat:PersistentScheduler--max-interval调度迭代之间最大睡眠秒数int对应max_interval-l, --loglevel日志级别默认WARNING-f, --logfile、--pidfile、--uid、--gid、--umask、--workdir继承自守护命令的通用选项此外还支持-C无彩色输出与额外配置参数透传ctx.args中未被识别的内容会交给app.config_from_cmdline(ctx.args)解析解析失败时抛出click.UsageError。8.3 启动流程与信号处理Beat.run()celery/apps/beat.py依次执行打印celery beat v{VERSION_BANNER} is starting.横幅init_loader()app.loader.init_worker()导入任务模块并app.finalize()设置进程标题celery beatstart_scheduler()若指定pidfile则platforms.create_pidlock构造Service并打印启动信息STARTUP_INFO_FMT展示 broker、loader、scheduler、logfile、maxinterval 等设置日志可选设置全局 socket 超时默认 30 秒安装同步信号处理器后service.start()。install_sync_handlercelery/apps/beat.py为SIGTERM与SIGINT安装处理器收到信号时先service.sync()保存调度表再抛SystemExit——确保关闭时调度状态如last_run_at、total_run_count不丢失。九、信号钩子celery.beat相关信号定义于 celery/signals.pybeat_init在Service.start()中、主循环开始前发送signals.beat_init.send(senderself)可用于初始化资源。beat_embedded_init仅在嵌入式进程模式embedded_processTrue下发送用于区分独立进程与嵌入应用进程两种运行形态。十、扩展调度器与测试验证10.1 自定义调度器Service.get_scheduler的extension_namespacecelery.beat_schedulers表明第三方调度器例如将调度表存入数据库的django_celery_beat.schedulers:DatabaseScheduler通过注册到该命名空间即可用配置或-S选项引用。自定义调度器通常继承Scheduler覆写setup_schedule、sync、close与schedule属性即可参考PersistentScheduler的模式。10.2 测试支撑模块级行为在 t/unit/app/test_beat.py 中有完整覆盖可作为理解实现语义的活文档代表性用例包括惰性参数求值test_beat_lazy_func条目推进test_next、test_reducetick 分支test_due_tick、test_pending_tick、test_honors_max_interval、test_ticks_schedule_change、test_not_due_top_entry_is_rescheduled_behind_due_entry同步计数test_should_sync、test_sync_task_counter_resets_on_do_sync默认条目与合并test_install_default_entries、test_merge_inplace错过 cron 的截止期限补偿test_tick_dispatches_missed_cron_within_deadline_non_uniform等。结语celery.beat模块以条目ScheduleEntry 调度器Scheduler 服务Service三层结构将周期任务调度抽象为清晰可扩展的组件堆驱动的 tick 循环保证调度效率持久化与同步机制保证状态不丢失beat_schedule配置与celery beat命令提供开箱即用的使用入口而beat_init/beat_embedded_init信号与celery.beat_schedulers扩展命名空间则为其在框架内的定制留足空间。理解这一模块是深入使用与扩展 Celery 周期任务能力的关键一步。赞分享任务调度后端消息队列【免费下载链接】celeryDistributed Task Queue (development branch)项目地址https://gitcode.com/gh_mirrors/ce/celery点击查看免费下载相关推荐Celery 周期性任务Periodic Tasks完整指南beat 调度器、Crontab 与 Solar 调度实战Celery 周期性任务Periodic Tasks完整指南beat 调度器、Crontab 与 Solar 调度实战 导读 本文以 Celery 官方用任务调度后端消息队列雀魂数据分析终极指南如何用免费工具快速提升麻将水平雀魂数据分析终极指南如何用免费工具快速提升麻将水平 想要从雀魂麻将新手成长为高手吗雀魂牌谱屋amae koromo就是你需要的秘密武器这款完全免费的开CANNAscend人工智能任务调度celery beat 命令完全指南配置、调度器与源码级原理剖析celery beat 命令完全指南配置、调度器与源码级原理剖析 导读 celery beat 是 Celery 分布式任务队列内置的周期任务调度器负责任务调度后端消息队列上一篇YYText手势处理终极指南双击放大与长按操作详解下一篇DictaLM 2.0性能评测希伯来语各项NLP任务的表现分析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考