任务调度后端【免费下载链接】apschedulerTask scheduling library for Python项目地址https://gitcode.com/gh_mirrors/ap/apscheduler点击查看免费下载APSchedulerAdvanced Python Scheduler是 Python 生态中兼具任务调度与任务队列能力的库既可用于单进程内的轻量定时任务也能通过共享数据存储与事件代理扩展到跨多节点的分布式部署。本篇以仓库 README.rst 为骨架结合源码实现系统讲解其同步/异步双形态、四大内置触发机制、持久化数据存储与事件代理、并发上限、misfire 容错与抖动等核心特性帮助读者掌握从单进程示例到多节点高可用的完整落地路径。一、项目定位任务调度器与任务队列的合体根据 README.rstAPScheduler 首先是一个任务调度器 任务队列系统。即使你完全没有调度需求也可以只把它当作纯作业队列来用——这意味着任务的执行与触发被解耦你可以只排队、只执行而不用关心何时触发。它的伸缩性覆盖两个极端单进程简单场景一个进程内运行调度器即可满足开箱即用见 examples/standalone/sync_memory.py。多节点大规模部署多个调度器与 worker 共享同一个数据存储从而同时获得高可用HA与水平扩展horizontal scaling能力。这种共享设计背后是调度与执行分离的架构任务、调度计划Schedule与作业Job都持久化在共享数据存储中任何一个节点都可以接替另一个节点的工作。同步与异步两种形态APScheduler 同时提供同步与异步两种使用形态覆盖传统线程型应用和 asyncio/Trio 异步应用同步形态Scheduler位于 src/apscheduler/_schedulers/sync.py。从源码看它是一个围绕AsyncScheduler的同步包装器启动时在一个独立线程中运行异步事件循环并在调度器关闭时一并关闭该线程。也就是说同步版本底层仍是异步事件循环只是对调用方隐藏了细节。异步形态AsyncScheduler位于 src/apscheduler/schedulers/async.py基于 AnyIO 实现需要 asyncio 或 Trio 作为底层事件循环。其文档注释还特别提示如果运行在 Trio 上要确保数据存储与事件代理与 Trio 兼容。两者统一从 src/apscheduler/init.py 对外导出用户只需from apscheduler import Scheduler或from apscheduler import AsyncScheduler即可。与此对应examples/standalone 目录中同时提供了sync_memory.py与async_memory.py两个最小示例展示了同一套 API 在同步与异步环境下的用法。Web 应用集成README 声明提供了与 WSGI 或 ASGI 兼容 Web 应用集成的文档与示例。仓库 examples/web 目录中给出了五种集成示例WSGI 路线wsgi_flask.pyFlask 框架、wsgi_noframework.py无框架原生 WSGIASGI 路线asgi_fastapi.pyFastAPI、asgi_starlette.pyStarlette、asgi_noframework.py无框架原生 ASGI二、持久化数据存储调度与作业的共享底座README 明确支持调度计划与作业的持久化存储这意味着它们可以在多个调度器/worker 实例间共享即使进程或节点重启调度计划与作业依然存在。内置数据存储后端内置的持久化数据存储后端共四类见 README.rst后端仓库实现PostgreSQL由 SQLAlchemy 数据存储统一覆盖MySQL 及其衍生品由 SQLAlchemy 数据存储统一覆盖SQLite由 SQLAlchemy 数据存储统一覆盖MongoDBsrc/apscheduler/datastores/mongodb.py其中 PostgreSQL/MySQL/SQLite 三者均由 src/apscheduler/datastores/sqlalchemy.py 中的SQLAlchemyDataStore实现——SQLAlchemy 的方言机制天然支持这三种数据库乃至更多。从源码看该数据存储在列类型层面做了大量兼容处理例如EmulatedTimestampTZ用 Unicode 字符串ISO 格式模拟带时区的时间戳src/apscheduler/datastores/sqlalchemy.pyEmulatedInterval用 BigInteger 模拟时间间隔从而在缺乏原生类型的数据库上保持行为一致。除持久化后端外还有基于内存的 src/apscheduler/datastores/memory.pyMemoryDataStore适合单进程、无需持久化的场景——它也是AsyncScheduler的默认数据存储见 src/apscheduler/schedulers/async.py。数据存储的用法示例在 examples/separate_worker/sync_scheduler.py 中可以看到真实用法用create_async_engine(postgresqlasyncpg://...)创建异步引擎然后传给SQLAlchemyDataStore(engine)即可作为调度器的data_store参数。README 也提到schedule 与 job 会持久化并跨节点存活。from sqlalchemy.ext.asyncio import create_async_engine from apscheduler import Scheduler from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker from apscheduler.triggers.interval import IntervalTrigger engine create_async_engine(postgresqlasyncpg://postgres:secretlocalhost/testdb) data_store SQLAlchemyDataStore(engine) event_broker AsyncpgEventBroker.from_async_sqla_engine(engine) with Scheduler(data_store, event_broker) as scheduler: scheduler.add_schedule(tick, IntervalTrigger(seconds1), idtick)注意该示例的核心思想是只写调度、不启动调度器——with块内没有调用run_until_stopped()调度器仅作为写入系统接口使用真正的执行交由另一端的 worker 进程完成见 examples/separate_worker/sync_worker.py 与async_worker.py。三、事件代理多节点协同的通知中枢当存在多个调度器/worker 时需要事件代理event broker来广播事件作业被添加、调度计划被更新、作业被获取/释放等。内置事件代理README 列出的内置事件代理有三个README.rst事件代理仓库实现依赖PostgreSQLsrc/apscheduler/eventbrokers/asyncpg.py、src/apscheduler/eventbrokers/psycopg.pyasyncpg / psycopgRedissrc/apscheduler/eventbrokers/redis.pyredisMQTTsrc/apscheduler/eventbrokers/mqtt.pypaho-mqtt此外还有进程内直接使用的 src/apscheduler/eventbrokers/local.pyLocalEventBroker它是AsyncScheduler的默认事件代理见 src/apscheduler/schedulers/async.py适合单进程场景。Redis 事件代理的源码细节以RedisEventBroker为例src/apscheduler/eventbrokers/redis.py接受一个异步 Redis 客户端或 Redis URL 字符串作为client_or_urlchannel参数指定消息广播频道默认值为apschedulerstop_check_interval控制频道监听器检查是否需要停止的间隔单位秒值越大响应停止越慢但 CPU 占用越低默认 1 秒若传入 URL 字符串代理会自行创建连接池并在退出时自动关闭_close_on_exit True若传入现成客户端实例代理不管理该客户端的生命周期需由使用方自行关闭——这是官方注释中明确提示的注意事项。启动时start方法代理会订阅指定频道并启动一个名为 Redis subscriber 的后台任务持续监听消息src/apscheduler/eventbrokers/redis.py。所有外部代理都基于tenacity对连接失败进行自动重试见_retry方法及 src/apscheduler/_retry.py 中的RetrySettings。从数据存储共享到事件广播的完整链路多个调度器/worker 共享同一数据存储因此任何节点都能看到全局的调度计划与作业在此基础上节点之间通过事件代理实时感知谁加了新调度、谁取走了作业、谁释放了作业从而避免重复执行并实现任务分配。这一模式对应 examples/separate_worker 目录sync_scheduler.py/async_scheduler.py只负责登记调度sync_worker.py/async_worker.py负责实际执行任务。四、内置调度机制触发器README 将内置调度机制统称为触发器trigger共四种README.rst触发器说明仓库实现Cron 风格调度仿照 UNIX cronsrc/apscheduler/triggers/cron/init.py间隔调度以固定间隔运行src/apscheduler/triggers/interval.py日历调度每隔 X 年/月/周/天触发且总是在一天中的同一时刻src/apscheduler/triggers/calendarinterval.py一次性调度在指定日期/时间运行一次src/apscheduler/triggers/date.pyCron 触发器CronTrigger的构造参数覆盖year/month/day/week/day_of_week/hour/minute/second全套时间字段并提供start_time最早触发时间默认当前时间、end_time最晚触发时间与timezone时区默认本地时区三个边界控制参数src/apscheduler/triggers/cron/init.py。其内部字段顺序由FIELDS_MAP明确定义src/apscheduler/triggers/cron/init.py字段表达式解析逻辑在 src/apscheduler/triggers/cron/expressions.py 中实现。需要注意的语义细节一周的第一天固定是周一README 与源码注释一致强调day_of_week支持 0-7 的数字0 与 7 均为周日或sun,mon,...名称支持标准的 crontab 五段表达式CronTrigger.from_crontab(expr)可把0 8 * * *这类表达式转换为触发器表达式必须恰好 5 段否则抛出ValueErrorsrc/apscheduler/triggers/cron/init.py。间隔触发器IntervalTrigger接受weeks/days/hours/minutes/seconds/microseconds六个单位参数首次触发时间为start_time创建时刻除非显式指定若给出end_time最后一次触发将不晚于该时刻src/apscheduler/triggers/interval.py。源码中有两个值得注意的约束src/apscheduler/triggers/interval.py总间隔必须为正数否则抛出ValueError(The time interval must be positive)end_time不得早于start_time否则抛出ValueError。最简单的可用示例就是 examples/standalone/sync_memory.py 中的IntervalTrigger(seconds1)——每秒触发一次tick()并打印当前时间。日历间隔触发器CalendarIntervalTrigger实现每隔 X 年/月/周/天触发、总是在一天中的同一时刻的日历语义如每个月 1 号 09:00。它区别于IntervalTrigger的关键在于按日历单位月、年长度不一计算而不是固定的时间增量。其实现位于 src/apscheduler/triggers/calendarinterval.py在 tests/triggers/test_calendarinterval.py 中有完整的测试覆盖。一次性触发器DateTrigger只需一个run_time参数仅在给定时刻触发一次。源码实现非常直白首次调用next()返回run_time并置_completed True此后一律返回Nonesrc/apscheduler/triggers/date.py。组合触发器README 特别指出不同的调度机制还可以通过所谓的组合触发器combining triggers结合使用。仓库中 src/apscheduler/triggers/combining.py 提供了两个组合器AndTrigger当多个触发器的触发时刻落在给定阈值threshold默认 1 秒内时返回最早的那个触发时刻否则让最早触发的子触发器继续前进再重新比对并受max_iterations默认 10000保护以防无限循环src/apscheduler/triggers/combining.py。OrTrigger返回任一子触发器产生的下一个触发时刻中最早的那个。组合触发器的使用在 tests/triggers/test_combining.py 中有系统性的测试验证。自定义触发器README 明确表示你可以通过构建自己的触发器类来实现自定义调度逻辑自定义触发器与内置触发器地位完全平等不会被区别对待。这得益于抽象基类 src/apscheduler/abc.py 中定义的Trigger接口核心方法是next()返回下一个触发时间。只要实现了该接口并处理好序列化所需的__getstate__/__setstate__如各内置触发器所示就可以自由接入调度器。五、其他值得注意的特性README 列举了三项对生产环境极具价值的高级特性README.rst1. 限制同一任务的最大并发作业数可以为给定任务函数限制同时运行的作业数量上限。源码中对应TaskDefaults.max_running_jobsAsyncScheduler在初始化时会为其设置默认值 1src/apscheduler/schedulers/async.py即默认情况下同一任务不会并发运行——这避免了同一任务的重入竞争。2. 限制作业允许延迟启动的时间可以为作业设置 misfire grace time误触发宽限时间限制一个作业被允许延迟多久才开始执行。若实际触发时间晚于计划时间且超出宽限该次执行将被放弃。源码中对应TaskDefaults.misfire_grace_time默认值为None表示不限制src/apscheduler/schedulers/async.py同时配套CoalescePolicy/ConflictPolicy等枚举见 src/apscheduler/_enums.py处理同一时刻多次触发的合并策略与冲突策略。3. Jitter抖动可为每个计划的作业运行时间添加可调节的随机延迟。这在多实例部署时尤其有用随机抖动能打散各节点在同一时刻的触发避免惊群效应式的资源争抢。此外从调度器构造参数src/apscheduler/schedulers/async.py还可看到更多生产级配置项max_concurrent_jobs全局并发作业上限默认 100、cleanup_interval自动清理过期数据的间隔默认 15 分钟、lease_duration对调度计划/任务上锁的最长持有时间默认 30 秒、role指定调度器在运行时只调度、只运行作业、或两者兼做等。六、版本状态与生产使用前提README 开篇即给出重要警告README.rstv4.0 系列目前是预发布pre-release版本可能以不向后兼容的方式变化且没有迁移路径请勿在生产环境中使用该版本。与之对应pyproject.toml 中声明的核心依赖为anyio 4.0、attrs、tenacity、tzlocal并要求 Python 3.10分类器中列出了 3.10 至 3.14 的支持范围。附加依赖按后端拆分asyncpg、cbor、mongodb、mqtt、redis、sqlalchemy等以可选依赖extras形式提供pyproject.toml用户可按需安装pip install APScheduler # 核心 pip install APScheduler[redis] # 启用 Redis 事件代理 pip install APScheduler[mongodb] # 启用 MongoDB 数据存储 pip install APScheduler[sqlalchemy] # 启用 SQLAlchemy 数据存储 pip install APScheduler[mqtt] # 启用 MQTT 事件代理七、快速上手同步内存版最小示例把以上概念落到最小可运行代码上仓库自带的 examples/standalone/sync_memory.py 就是最直观的入门样例from datetime import datetime from apscheduler import Scheduler from apscheduler.triggers.interval import IntervalTrigger def tick(): print(Hello, the time is, datetime.now()) with Scheduler() as scheduler: scheduler.add_schedule(tick, IntervalTrigger(seconds1)) scheduler.run_until_stopped()运行python sync_memory.py控制台会以每秒一次的频率打印当前时间。这个例子虽然简单却完整展示了 APScheduler 的四个核心元素任务Task即tick函数由add_schedule自动注册调度计划Schedule任务 触发器IntervalTrigger(seconds1)的绑定数据存储默认的MemoryDataStore事件代理默认的LocalEventBroker。默认情况下Scheduler()不传参调度器内部会自动装配三个作业执行器src/apscheduler/schedulers/async.pyasync异步执行器、threadpool线程池执行器、processpool进程池执行器对应 src/apscheduler/executors 目录下的实现同步形态的Scheduler默认任务执行器为threadpoolsrc/apscheduler/_schedulers/sync.py。结语APScheduler 的设计在简单与规模之间取得了平衡单进程内开箱即用的内存数据存储与本地事件代理与面向多节点的 PostgreSQL/MySQL/SQLite/MongoDB 持久化 PostgreSQL/Redis/MQTT 事件广播共用同一套触发器与调度 API。理解数据存储与事件代理的职责边界、四种触发器及其组合/自定义机制、以及并发上限、misfire 宽限与 jitter 三项生产特性是正确使用并进一步扩展这套系统的关键。仓库中的 examples 目录提供了从独立内存版到分离 worker 版、再到 Web 集成版的完整进阶路线可作为动手实践的参考。赞分享任务调度后端【免费下载链接】apschedulerTask scheduling library for Python项目地址https://gitcode.com/gh_mirrors/ap/apscheduler点击查看免费下载相关推荐TDengine 3.x 分布式架构深度解析逻辑节点、存储模型与数据复制机制TDengine 3.x 分布式架构深度解析逻辑节点、存储模型与数据复制机制 TDengine 从诞生之初就采用分布式 高可用 水平扩展的设计理念数据库时序数据库物联网大数据实时分析云原生Sketch Constraints与原生约束对比哪款布局工具更适合你的设计需求Sketch Constraints与原生约束对比哪款布局工具更适合你的设计需求 在UI设计中高效的布局工具是提升工作效率的关键。Sketch作为主流设计CANN/asc-devkitSIMT-API half2对数函数h2log 产品支持情况 ! npu950 id1 Ascend 950PR/Ascend 950DT支持 ! end id1 ! npuA3人工智能深度学习算子库CANNAscend上一篇如何通过Conventional Commits提升测试驱动开发效率5个实用技巧下一篇CANN/asc-devkit向量最大值标量计算创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考