本文介绍了 Python 中实现定时任务的多种方案，包括基础循环、标准库模块及第三方框架。主要方法涵盖 while 循环配合 sleep 函数、threading.Timer 和 sched 模块、轻量级 schedule 库、功能完善的 APScheduler 框架，以及用于分布式调度的 Celery 和数据流管理工具 Apache Airflow。不同方案在阻塞性、并发能力、持久化支持及依赖管理上各有优劣，开发者应根据具体业务场景选择合适工具。
在日常工作中，我们常常会用到需要周期性执行的任务。一种方式是采用 Linux 系统自带的 crond 结合命令行实现，另外一种方式是直接使用 Python。接下来整理的是常见的 Python 定时任务的实现方式。
位于 time 模块中的 sleep(secs) 函数，可以实现令当前执行的线程暂停 secs 秒后再继续执行。所谓暂停，即令当前线程进入阻塞状态，当达到 sleep() 函数规定的时间后，再由阻塞状态转为就绪状态，等待 CPU 调度。
基于这样的特性我们可以通过 while 死循环 +sleep() 的方式实现简单的定时任务。
代码示例：
import datetime
import time
def time_printer():
now = datetime.datetime.now()
ts = now.strftime('%Y-%m-%d %H:%M:%S')
print('do func time :', ts)
def loop_monitor():
while True:
time_printer()
time.sleep(5) # 暂停 5 秒
if __name__ == "__main__":
loop_monitor()
主要缺点：
Timeloop 是一个库，可用于运行多周期任务。这是一个简单的库，它使用 decorator 模式在线程中运行标记函数。
示例代码：
import time
from timeloop import Timeloop
from datetime import timedelta
tl = Timeloop()
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
print("2s job current time : {}".format(time.ctime()))
@tl.job(interval=timedelta(seconds=5))
def sample_job_every_5s():
print("5s job current time : {}".format(time.ctime()))
@tl.job(interval=timedelta(seconds=10))
def sample_job_every_10s():
print("10s job current time : {}".format(time.ctime()))
threading 模块中的 Timer 是一个非阻塞函数，比 sleep 稍好一点。Timer 最基本理解就是定时器，我们可以启动多个定时任务，这些定时器任务是异步执行，所以不存在等待顺序执行问题。
Timer(interval, function, args=[], kwargs={})
代码示例：
import threading
import time
def task():
print(f"Task executed at {time.strftime('%H:%M:%S')}")
# 创建定时器，5 秒后执行 task
t = threading.Timer(5, task)
t.start()
备注：Timer 只能执行一次，如果需要循环调用，需要在任务内部再次创建或循环调用。
sched 模块实现了一个通用事件调度器，在调度器类使用一个延迟函数等待特定的时间，执行任务。同时支持多线程应用程序，在每个任务执行后会立刻调用延时函数，以确保其他线程也能执行。
class sched.scheduler(timefunc, delayfunc) 这个类定义了调度事件的通用接口，它需要外部传入两个参数，timefunc 是一个没有参数的返回时间类型数字的函数 (常用使用的如 time 模块里面的 time)，delayfunc 应该是一个需要一个参数来调用、与 timefunc 的输出兼容、并且作用为延迟多个时间单位的函数 (常用的如 time 模块的 sleep)。
代码示例：
import datetime
import time
import sched
def time_printer():
now = datetime.datetime.now()
ts = now.strftime('%Y-%m-%d %H:%M:%S')
print('do func time :', ts)
def loop_monitor():
s = sched.scheduler(time.time, time.sleep) # 生成调度器
s.enter(5, 1, time_printer, ())
s.run()
if __name__ == "__main__":
loop_monitor()
scheduler 对象主要方法:
个人点评：比 threading.Timer 更好，不需要循环调用。
schedule 是一个第三方轻量级的任务调度模块，可以按照秒，分，小时，日期或者自定义事件执行时间。schedule 允许用户使用简单、人性化的语法以预定的时间间隔定期运行 Python 函数 (或其它可调用函数)。
先来看代码，是不是不看文档就能明白什么意思？
import schedule
import time
def job():
print("I'm working...")
schedule.every(10).seconds.do(job)
schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)
while True:
schedule.run_pending()
time.sleep(1)
装饰器：通过 @repeat() 装饰静态方法
import time
from schedule import every, repeat, run_pending
@repeat(every().second)
def job():
print('working...')
while True:
run_pending()
time.sleep(1)
传递参数：
import schedule
def greet(name):
print('Hello', name)
schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')
while True:
schedule.run_pending()
取消任务：
import schedule
i = 0
def some_task():
global i
i += 1
print(i)
if i == 10:
schedule.cancel_job(job)
print('cancel job')
exit(0)
job = schedule.every().second.do(some_task)
while True:
schedule.run_pending()
根据标签检索任务：
import schedule
def greet(name):
print('Hello {}'.format(name))
schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')
friends = schedule.get_jobs('friend')
print(friends)
并行运行：使用 Python 内置队列实现：
import threading
import time
import schedule
def job1():
print("I'm running on thread %s" % threading.current_thread())
def job2():
print("I'm running on thread %s" % threading.current_thread())
def job3():
print("I'm running on thread %s" % threading.current_thread())
def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()
schedule.every(10).seconds.do(run_threaded, job1)
schedule.every(10).seconds.do(run_threaded, job2)
schedule.every(10).seconds.do(run_threaded, job3)
while True:
schedule.run_pending()
time.sleep(1)
APScheduler (advanced python scheduler) 基于 Quartz 的一个 Python 定时任务框架，实现了 Quartz 的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及 crontab 类型的任务，并且可以持久化任务。基于这些功能，我们可以很方便的实现一个 Python 定时任务系统。
它有以下三个特点：
APScheduler 有四种组成部分：
示例代码：
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 输出时间
def job():
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5, id='my_job_id')
sched.start()
Job 作为 APScheduler 最小执行单位。创建 Job 时指定执行的函数，函数中所需参数，Job 执行时的一些设置信息。
构建说明：
Trigger 绑定到 Job，在 scheduler 调度筛选 Job 时，根据触发器的规则计算出 Job 的触发时间，然后与当前时间比较确定此 Job 是否会被执行，总之就是根据 trigger 规则计算出下一个执行时间。
目前 APScheduler 支持触发器：
触发器参数：date date 定时，作业只执行一次。 run_date (datetime|str) – the date/time to run the job at timezone (datetime.tzinfo|str) – time zone for run_date if it doesn't have one already
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
sched.add_job(my_job, 'date', run_date=datetime(2019, 7, 6, 16, 30, 5), args=['text'])
触发器参数：interval interval 间隔调度 weeks (int) – 间隔几天 days (int) – 间隔几天 hours (int) – 间隔几小时 minutes (int) – 间隔几分钟 seconds (int) – 间隔多少秒 start_date (datetime|str) – 开始日期 end_date (datetime|str) – 结束日期 timezone (datetime.tzinfo|str) – 时区
sched.add_job(job_function, 'interval', hours=2)
触发器参数：cron cron 调度 (int|str) 表示参数既可以是 int 类型，也可以是 str 类型 (datetime | str) 表示参数既可以是 datetime 类型，也可以是 str 类型 year (int|str) – 4-digit year -（表示四位数的年份，如 2008 年） month (int|str) – month (1-12) -（表示取值范围为 1-12 月） day (int|str) – day of the (1-31) -（表示取值范围为 1-31 日） week (int|str) – ISO week (1-53) -（格里历 2006 年 12 月 31 日可以写成 2006-W52-7（扩展形式）或 2006W527（紧凑形式）） day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) – （表示一周中的第几天，既可以用 0-6 表示也可以用其英语缩写表示） hour (int|str) – hour (0-23) – （表示取值范围为 0-23 时） minute (int|str) – minute (0-59) – （表示取值范围为 0-59 分） second (int|str) – second (0-59) – （表示取值范围为 0-59 秒） start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) – （表示开始时间） end_date (datetime|str) – latest possible date/time to trigger on (inclusive) – （表示结束时间） timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) -（表示时区取值）
CronTrigger 可用的表达式：
| 表达式 | 参数类型 | 描述 |
|---|---|---|
| * | 所有 | 通配符。例：minutes=*即每分钟触发 |
| * / a | 所有 | 每隔时长 a 执行一次。例：minutes="* / 3" 即每隔 3 分钟执行一次 |
| a – b | 所有 | a – b 的范围内触发。例：minutes="2-5"。即 2 到 5 分钟内每分钟执行一次 |
| a – b / c | 所有 | a – b 范围内，每隔时长 c 执行一次。 |
| xth y | 日 | 第几个星期几触发。x 为第几个，y 为星期几 |
| last x | 日 | 一个月中，最后一个星期的星期几触发 |
| last | 日 | 一个月中的最后一天触发 |
| x, y, z | 所有 | 组合表达式，可以组合确定值或上述表达式 |
# 6-8,11-12 月第三个周五 00:00, 01:00, 02:00, 03:00 运行
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# 每周一到周五运行 直到 2024-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2024-05-30')
Executor 在 scheduler 中初始化，另外也可通过 scheduler 的 add_executor 动态添加 Executor。每个 executor 都会绑定一个 alias，这个作为唯一标识绑定到 Job，在实际执行时会根据 Job 绑定的 executor 找到实际的执行器对象，然后根据执行器对象执行 Job。
Executor 的种类会根据不同的调度来选择，如果选择 AsyncIO 作为调度的库，那么选择 AsyncIOExecutor，如果选择 tornado 作为调度的库，选择 TornadoExecutor，如果选择启动进程作为调度，选择 ThreadPoolExecutor 或者 ProcessPoolExecutor 都可以。
Executor 的选择需要根据实际的 scheduler 来选择不同的执行器。目前 APScheduler 支持的 Executor：
Jobstore 在 scheduler 中初始化，另外也可通过 scheduler 的 add_jobstore 动态添加 Jobstore。每个 jobstore 都会绑定一个 alias，scheduler 在 Add Job 时，根据指定的 jobstore 在 scheduler 中找到相应的 jobstore，并将 job 添加到 jobstore 中。作业存储器决定任务的保存方式，默认存储在内存中（MemoryJobStore），重启后就没有了。APScheduler 支持的任务存储器有：
不同的任务存储器可以在调度器的配置中进行配置（见调度器）
Event 是 APScheduler 在进行某些操作时触发相应的事件，用户可以自定义一些函数来监听这些事件，当触发某些 Event 时，做一些具体的操作。常见的比如。Job 执行异常事件 EVENT_JOB_ERROR。Job 执行时间错过事件 EVENT_JOB_MISSED。
目前 APScheduler 定义的 Event： EVENT_SCHEDULER_STARTED, EVENT_SCHEDULER_START, EVENT_SCHEDULER_SHUTDOWN, EVENT_SCHEDULER_PAUSED, EVENT_SCHEDULER_RESUMED, EVENT_EXECUTOR_ADDED, EVENT_EXECUTOR_REMOVED, EVENT_JOBSTORE_ADDED, EVENT_JOBSTORE_REMOVED, EVENT_ALL_JOBS_REMOVED, EVENT_JOB_ADDED, EVENT_JOB_REMOVED, EVENT_JOB_MODIFIED, EVENT_JOB_EXECUTED, EVENT_JOB_ERROR, EVENT_JOB_MISSED, EVENT_JOB_SUBMITTED, EVENT_JOB_MAX_INSTANCES。
Listener 表示用户自定义监听的一些 Event，比如当 Job 触发了 EVENT_JOB_MISSED 事件时可以根据需求做一些其他处理。
Scheduler 是 APScheduler 的核心，所有相关组件通过其定义。scheduler 启动之后，将开始按照配置的任务进行调度。除了依据所有定义 Job 的 trigger 生成的将要调度时间唤醒调度之外。当发生 Job 信息变更时也会触发调度。
APScheduler 支持的调度器方式如下，比较常用的为 BlockingScheduler 和 BackgroundScheduler
Celery 是一个简单，灵活，可靠的分布式系统，用于处理大量消息，同时为操作提供维护此类系统所需的工具，也可用于任务调度。Celery 的配置比较麻烦，如果你只是需要一个轻量级的调度工具，Celery 不会是一个好选择。
Celery 是一个强大的分布式任务队列，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。异步任务比如是发送邮件、或者文件上传，图像处理等等一些比较耗时的操作，定时任务是需要在特定时间执行的任务。
需要注意，celery 本身并不具备任务的存储功能，在调度任务的时候肯定是要把任务存起来的，因此在使用 celery 的时候还需要搭配一些具备存储、访问功能的工具，比如：消息队列、Redis 缓存、数据库等。官方推荐的是消息队列 RabbitMQ，有些时候使用 Redis 也是不错的选择。
它的架构组成如下图：
Celery 架构，它采用典型的生产者 - 消费者模式，主要由以下部分组成：
实际应用中，用户从 Web 前端发起一个请求，我们只需要将请求所要处理的任务丢入任务队列 broker 中，由空闲的 worker 去处理任务即可，处理的结果会暂存在后台数据库 backend 中。我们可以在一台机器或多台机器上同时起多个 worker 进程来实现分布式地并行处理任务。
Apache Airflow 是 Airbnb 开源的一款数据流程工具，目前是 Apache 孵化项目。以非常灵活的方式来支持数据的 ETL 过程，同时还支持非常多的插件来完成诸如 HDFS 监控、邮件通知等功能。Airflow 支持单机和分布式两种模式，支持 Master-Slave 模式，支持 Mesos 等资源调度，有非常好的扩展性。被大量公司采用。
Airflow 使用 Python 开发，它通过 DAGs(Directed Acyclic Graph, 有向无环图) 来表达一个工作流中所要执行的任务，以及任务之间的关系和依赖。比如，如下的工作流中，任务 T1 执行完成，T2 和 T3 才能开始执行，T2 和 T3 都执行完成，T4 才能开始执行。
Airflow 提供了各种 Operator 实现，可以完成各种任务实现：
除了以上这些 Operators 还可以方便的自定义 Operators 满足个性化的任务需求。
一些情况下，我们需要根据执行结果执行不同的任务，这样工作流会产生分支。这种需求可以使用 BranchPythonOperator 来实现。
通常，在一个运维系统，数据分析系统，或测试系统等大型系统中，我们会有各种各样的依赖需求。包括但不限于：
crontab 可以很好地处理定时执行任务的需求，但仅能管理时间上的依赖。Airflow 的核心概念 DAG（有向无环图）—— 来表现工作流。
Airflow 是一种 WMS，即：它将任务以及它们的依赖看作代码，按照那些计划规范任务执行，并在实际工作进程之间分发需执行的任务。
Airflow 提供了一个用于显示当前活动任务和过去任务状态的优秀 UI，并允许用户手动管理任务的执行和状态。
Airflow 中的工作流是具有方向性依赖的任务集合。
DAG 中的每个节点都是一个任务，DAG 中的边表示的是任务之间的依赖（强制为有向无环，因此不会出现循环依赖，从而导致无限执行循环）。
在一个可扩展的生产环境中，Airflow 含有以下组件：
Worker 的具体实现由配置文件中的 executor 来指定，airflow 支持多种 Executor:
生产环境一般使用 CeleryExecutor 和 KubernetesExecutor。
以上总结了 Python 定时任务的常用方案，开发者可根据实际需求选择：
选择合适的工具可以提高开发效率并确保系统的稳定性。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online