python定时任务

发表于 更新于 分类于 阅读次数：

在项目中，我们可能会遇到定时任务，比如要每隔一段时间执行一次任务，并且还需要在不打断循环的基础上修改任务。这时，我们可以使用 APScheduler 框架。

APScheduler 的使用

APScheduler 是一个轻量级的 Python 定时任务调度框架。APScheduler 支持三种调度任务：固定时间间隔，固定时间点（日期），Linux 下的 Crontab 命令。同时，它还支持异步执行、后台执行调度任务。

安装
1
pip install APScheduler
使用

可以概括为一下三个步骤：

  1. 新建一个调度器（schedulers）
  2. 添加一个调度任务
  3. 运行调度任务

接下来我们看一个示例：

1
2
3
4
5
6
7
8
9
10
11
12
from apscheduler.schedulers.background import BackgroundScheduler

def count(n):
    print(n)

if __name__ == '__main__':
    # 创建一个调度器
    scheduler = BackgroundScheduler()
    # 添加调度任务
    scheduler.add_job(id='1', func=count, trigger='interval', seconds=5, args=(3,))
    # 开始调度
    scheduler.start()

add_job 函数参数解释：

  • id：任务的 id，用于区分不同的任务
  • func：任务函数，可直接传入回调函数，或者以 package.module:some.object（函数所在路径） 的形式传入
  • trigger：触发器，’interval’ ，’date’， ‘cron’ 三种可选
  • seconds：当触发器为 interval 时传入，表示时间间隔
  • args：任务函数的参数

对定时任务有了初步了解后，我们来介绍一下 APScheduler 的几个核心组件。

调度器

APScheduler 提供了七种调度器：

  • BlockingScheduler : 调度器在当前进程的主线程中运行，也就是会阻塞当前线程。
  • BackgroundScheduler : 调度器在后台线程中运行，不会阻塞当前线程（常用）。
  • AsyncIOScheduler : 结合 asyncio 模块（一个异步框架）一起使用。
  • GeventScheduler : 程序中使用 gevent（高性能的Python并发框架）作为IO模型，和 GeventExecutor 配合使用。
  • TornadoScheduler : 程序中使用 Tornado（一个web框架）的IO模型，用 ioloop.add_timeout 完成定时唤醒。
  • TwistedScheduler : 配合 TwistedExecutor，用 reactor.callLater 完成定时唤醒。
  • QtScheduler : 你的应用是一个 Qt 应用，需使用QTimer完成定时唤醒。
触发器
1. date

作业任务只会执行一次，它表示特定的时间点触发。当触发器类型为 date 时 add_job 的参数如下：

参数 说明
run_date (datetime 或 str) 作业的运行日期或时间
timezone (datetime.tzinfo 或 str) 指定时区

示例如下：

1
2
3
4
5
6
7
from datetime import date, datetime

scheduler.add_job(id='1', func=count, trigger='date', run_date=date(2022, 1, 1), args=(3,))

scheduler.add_job(id='1', func=count, trigger='date', run_date=datetime(2022, 1, 1, 14, 30, 20), args=(3,))

scheduler.add_job(id='1', func=count, trigger='date', run_date='2022-1-1 14:30:20', args=(3,))

2. interval

固定时间间隔触发，循环执行。当触发器类型为 date 时 add_job 的参数如下：

参数 说明
weeks (int) 间隔几周
days (int) 间隔几天
hours (int) 间隔几小时
minutes (int) 间隔几分钟
seconds (int) 间隔几秒
start_date (datetime 或 str) 开始日期
end_date (datetime 或 str) 结束日期
timezone (datetime.tzinfo 或str) 时区
作业存储
添加 job

有两种添加任务的方法，第一种就是上面用到的 add_job 函数，另一种是通过修饰器 scheduled_job() 来添加任务。

1
2
3
@scheduler.scheduled_job(id='1', trigger='interval', seconds=5)
def count(n):
    print(n)

不同的是：第一种方法返回一个 apscheduler.job.Job 实例，可以用来改变或者移除 job。第二种方法只适用于应用运行期间不会改变的 job。


移除 job

移除 job 也有两种方法：remove_job() 和 job.remove()。
remove_job() 是根据 job 的 id 来移除，所以要在 job 创建的时候指定一个 id。
job.remove() 则是对 job 执行 remove 方法即可。

1
2
3
4
scheduler.remove_job('1')

job = scheduler.add_job()
job.remove()

修改 job

可以在调度开始之后修改 job。可以使用 Job.modify() 或者 modify_job() 方法来修改 job 的属性。但是值得注意的是，job 的 id 是无法被修改的。

1
2
3
4
5
scheduler = BackgroundScheduler()

scheduler.add_job(id='1', func=count, trigger='interval', seconds=5, args=(3,))

scheduler.modify_job('1', trigger='interval', seconds=10, args=(3, 5))

关闭 job

默认情况下调度器会等待所有正在运行的作业完成后，关闭所有的调度器和作业存储。如果你不想等待，可以将 wait 选项设置为 False。

1
scheduler.shutdown(wait=False)

flask 中使用定时任务

flask-apscheduler 是 flask 框架的一个扩展，它对 apscheduler 进行了进一步封装。

安装
1
pip install flask-apscheduler
配置

在配置文件中添加如下内容：

1
JOBS = [    {        'id': '1',  # 一个标识        'func': 'app:tasks.task.count',  # 指定运行的函数        'args': (3,),  # 传入函数的参数        'trigger': 'interval',  # 指定 定时任务的类型        'seconds': 3  # 运行的间隔时间    }]SCHEDULER_API_ENABLED = True

然后在工厂函数中进行初始化：

1
from flask_apscheduler import APSchedulerfrom apscheduler.schedulers.background import BackgroundScheduler# 指明时区，避免在部署时因时区不统一报错scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))def create_app():    ...    scheduler.init_app(app)    scheduler.start()    ...

scheduler 是一个调度器，其用法与 apscheduler 一样。