Python 实现定时任务的八种方案！

阅读量：795 次

发布时间：2023-03-07

本文共 3926 字，大约阅读时间需要 13 分钟。

Python 定时任务实现指南

在日常开发中，定时任务是常见需求之一。Linux 系统提供了 crond 工具，但对于 Python 开发者来说，直接使用 Python 实现定时任务更为灵活。以下是几种常见的实现方式及其优缺点分析。

1. while True + sleep() 实现定时任务

基于 time.sleep() 函数，可以通过 while True 循环实现简单的定时任务。这种方法的核心原理是使用 sleep() 函数暂停当前线程。

优点：

简单易实现，适合快速开发。

缺点：

仅能设置间隔时间，无法指定具体的执行时间（如每天8点）。

sleep() 是阻塞函数，会阻塞当前线程，导致资源浪费。

示例代码：

import timedef time_printer():    now = datetime.datetime.now()    ts = now.strftime('%Y-%m-%d %H:%M:%S')    print(f'do func time: {ts}')def loop_monitor():    while True:        time_printer()        time.sleep(5)  # 暂停5秒if __name__ == "__main__":    loop_monitor()

2. 使用 Timeloop 库运行定时任务

Timeloop 是一个轻量级库，支持在线程中运行多周期任务。通过装饰器模式，可以标记需要定期执行的函数。

优点：

简单易用，支持多个周期任务。

使用 @timeloop.job 装饰器，语法直观。

示例代码：

import timefrom timeloop import Timelooptl = Timeloop()@tl.job(interval=time.timedelta(seconds=2))def sample_job_every_2s():    print("2s job current time: {}".format(time.ctime()))@tl.job(interval=time.timedelta(seconds=5))def sample_job_every_5s():    print("5s job current time: {}".format(time.ctime()))@tl.job(interval=time.timedelta(seconds=10))def sample_job_every_10s():    print("10s job current time: {}".format(time.ctime())

3. 使用 threading.Timer 实现定时任务

threading.Timer 是一个非阻塞定时器，可以在多个线程中运行定时任务。

优点：

非阻塞，避免了线程资源的浪费。

支持异步执行，任务间不会互相干扰。

缺点：

Timer 只能执行一次任务，需要手动循环调用。

示例代码：

import threadingimport timedef timer_task():    while True:        print("I'm running on thread {}".format(threading.current_thread()))        time.sleep(1)def run_timer_task():    timer = threading.Timer(1, timer_task)    timer.start()if __name__ == "__main__":    run_timer_task()

4. 使用内置模块 sched 实现定时任务

sched 模块提供了一个通用事件调度器，支持多线程环境，能够在短时间内执行任务。

优点：

灵活性高，支持多种调度方式。

无需手动循环调用任务。

示例代码：

import timeimport scheddef time_printer():    now = datetime.datetime.now()    ts = now.strftime('%Y-%m-%d %H:%M:%S')    print(f'do func time: {ts}')# 创建调度器scheduler = sched.scheduler(time.time, time.sleep)scheduler.enter(5, 1, time_printer, ())  # 每5秒执行一次scheduler.run()if __name__ == "__main__":    scheduler.start()

5. 使用调度模块 schedule 实现定时任务

schedule 是一个轻量级的第三方模块，支持灵活的任务调度，包括秒、分钟、小时、日期等时间间隔。

优点：

语法简洁，支持丰富的调度方式。

支持传递参数和标签，方便任务管理。

示例代码：

import schedulefrom datetime import datetime, timedeltadef greet(name):    print(f'Hello {name}')# 每2秒执行一次schedule.every(2).seconds.do(greet, name='Alice')# 每4秒执行一次schedule.every(4).seconds.do(greet, name='Bob')while True:    schedule.run_pending()    time.sleep(1)

6. 使用任务框架 APScheduler 实现定时任务

APScheduler 基于 Quartz，提供了类似 Linux Cron 的调度功能，支持周期性和固定时间点的任务调度。

核心概念：

Job 作业：最小执行单位，包含执行函数和配置。

Trigger 触发器：根据规则计算下次执行时间。

Executor 执行器：处理作业的线程或进程池。

Jobstore 作业存储：存储作业的持久化存储。

Event 事件：用于监听作业状态变化。

示例代码：

from apscheduler.schedulers.blocking import BlockingSchedulerfrom datetime import datetimedef job():    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))scheduler = BlockingScheduler()scheduler.add_job(job, 'interval', seconds=5, id='my_job_id')scheduler.start()

7. 使用分布式消息系统 Celery 实现定时任务

Celery 是一个分布式任务队列系统，支持异步任务和定时任务。其架构包括 Beat（调度器）、生产者、消息中间件（Broker）、消费者和结果存储 backend。

核心组件：

Celery Beat：负责周期性发布任务。

RabbitMQ/Broker：消息中间件。

Worker：执行任务的消费者。

示例代码：

from celery import task@task(name='task_name')def my_task():    print('Task executed!')# 在 Broker 中添加任务with app.pools['worker'] as pool:    pool.put_task('task_name', (1, 2))

8. 使用数据流工具 Apache Airflow 实现定时任务

Airflow 是一个灵活的数据流工具，支持 DAG（有向无环图）定义工作流，适合复杂的任务依赖关系。

核心概念：

DAG：描述任务的执行顺序。

Operators：任务执行器，如 BashOperator、PythonOperator 等。

Task Instance：任务的执行实例。

示例代码：

from airflow import DAG, BranchPythonOperator, PythonOperatorfrom datetime import datetime, timedeltadefault_dag = DAG('example_dag', schedule_interval=timedelta(days=1))with default_dag:    start = BranchPythonOperator(        task_id='start',        python_callable=your_function,        op_kwargs={'param1': 1}    )    start.execute()