Python中如何优雅的使用定时任务？

妖狐艹你老母 2021-09-25 09:32 241阅读 0赞

> 在项目中，我们可能遇到有定时任务的需求。其一：定时执行任务。例如每天早上 8 点定时推送早报。其二：每隔一个时间段就执行任务。比如：每隔一个小时提醒自己起来走动走动，避免长时间坐着。今天，我跟大家分享下 Python 定时任务的实现方法。

# 1 #

第一种办法是最简单又最暴力。那就是在一个死循环中，使用线程睡眠函数 sleep()。

from datetime import datetime
    import time
    
    '''
    每个 10 秒打印当前时间。
    '''
    def timedTask():
        while True:
            print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
            time.sleep(10)
    
    if __name__ == '__main__':
        timedTask()

这种方法能够执行固定间隔时间的任务。如果`timedTask()`函数之后还有些操作，我们还使用死循环 + 阻塞线程。这会使得`timedTask()`一直占有 CPU 资源，导致后续操作无法执行。我建议谨重使用。

# 2 #

既然第一种方法暴力，那么有没有比较优雅地方法？答案是肯定的。Python 标准库 threading 中有个 Timer 类。它会新启动一个线程来执行定时任务，所以它是非阻塞函式。

如果你有使用多线程的话，需要关心线程安全问题。那么你可以选使用`threading.Timer`模块。

from datetime import datetime
    from threading import Timer
    import time
    
    '''
    每个 10 秒打印当前时间。
    '''
    def timedTask():
        '''
        第一个参数: 延迟多长时间执行任务(单位: 秒)
        第二个参数: 要执行的任务, 即函数
        第三个参数: 调用函数的参数(tuple)
        '''
        Timer(10, task, ()).start()
    
    # 定时任务
    def task():
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    if __name__ == '__main__':
        timedTask()
        while True:
            print(time.time())
            time.sleep(5)

运行结果：

1512486945.1196375
    1512486950.119873
    2017-12-05 23:15:50
    1512486955.133385

# 3 #

第三种方式是使用标准库中`sched`模块。sched 是事件调度器，它通过 `scheduler` 类来调度事件，从而达到定时执行任务的效果。

`sched`库使用起来也是非常简单。  
1）首先构造一个`sched.scheduler`类  
它接受两个参数：`timefunc` 和 `delayfunc`。timefunc 应该返回一个数字，代表当前时间，delayfunc 函数接受一个参数，用于暂停运行的时间单元。

一般使用默认参数就行，即传入这两个参数 `time.time` 和 `time.sleep`.当然，你也可以自己实现时间暂停的函数。

2）添加调度任务  
`scheduler` 提供了两个添加调度任务的函数:

*  `enter(delay, priority, action, argument=(), kwargs={})`

该函数可以延迟一定时间执行任务。`delay` 表示延迟多长时间执行任务，单位是秒。`priority`为优先级，越小优先级越大。两个任务指定相同的延迟时间，优先级大的任务会向被执行。`action` 即需要执行的函数，`argument` 和 `kwargs` 分别是函数的位置和关键字参数。

*  `scheduler.enterabs(time, priority, action, argument=(), kwargs={})`

添加一项任务，但这个任务会在 `time` 这时刻执行。因此，`time` 是绝对时间.其他参数用法与 `enter()` 中的参数用法是一致。

3）把任务运行起来  
调用 `scheduler.run()`函数就完事了。

下面是 sche 使用的简单示例：

from datetime import datetime
    import sched
    import time
    
    '''
    每个 10 秒打印当前时间。
    '''
    def timedTask():
        # 初始化 sched 模块的 scheduler 类
        scheduler = sched.scheduler(time.time, time.sleep)
        # 增加调度任务
        scheduler.enter(10, 1, task)
        # 运行任务
        scheduler.run()
    
    # 定时任务
    def task():
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    if __name__ == '__main__':
        timedTask()

值得注意的是： scheduler 中的每个调度任务只会工作一次，不会无限循环被调用。如果想重复执行同一任务， 需要重复添加调度任务即可。

通过上文，我们了解到有三种办法能实现定时任务，但是都无法做到循环执行定时任务。因此，需要一个能够担当此重任的库。它就是`APScheduler`。

# 1 简介 #

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

# 2 安装 #

使用 pip 包管理工具安装 APScheduler 是最方便快捷的。

pip install APScheduler
    # 如果出现因下载失败导致安装不上的情况，建议使用代理
    pip --proxy http://代理ip:端口 install APScheduler

# 3 使用步骤 #

APScheduler 使用起来还算是比较简单。运行一个调度任务只需要以下三部曲。

1.  新建一个 schedulers (调度器) 。
2.  添加一个调度任务(job stores)。
3.  运行调度任务。

下面是执行每 2 秒报时的简单示例代码：

import datetime
    import time
    from apscheduler.schedulers.background import BackgroundScheduler
    
    def timedTask():
        print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
    
    if __name__ == '__main__':
        # 创建后台执行的 schedulers
        scheduler = BackgroundScheduler()  
        # 添加调度任务
        # 调度方法为 timedTask，触发器选择 interval(间隔性)，间隔时长为 2 秒
        scheduler.add_job(timedTask, 'interval', seconds=2)
        # 启动调度任务
        scheduler.start()
        
        while True:
            print(time.time())
            time.sleep(5)

# 4 基础组件 #

APScheduler 有四种组件，分别是：`调度器(scheduler)`，`作业存储(job store)`，`触发器(trigger)`，`执行器(executor)`。

*  schedulers（调度器）  
    它是任务调度器，属于控制器角色。它配置作业存储器和执行器可以在调度器中完成，例如添加、修改和移除作业。
 *  triggers（触发器）  
    描述调度任务被触发的条件。不过触发器完全是无状态的。
 *  job stores（作业存储器）  
    任务持久化仓库，默认保存任务在内存中，也可将任务保存都各种数据库中，任务中的数据序列化后保存到持久化数据库，从数据库加载后又反序列化。
 *  executors（执行器）  
    负责处理作业的运行，它们通常通过在作业中提交指定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。

## 4.1 schedulers（调度器） ##

我个人觉得 APScheduler 非常好用的原因。它提供 7 种调度器，能够满足我们各种场景的需要。例如：后台执行某个操作，异步执行操作等。调度器分别是：

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

## 4.2 triggers（触发器） ##

APScheduler 有三种内建的 trigger:  
1）date 触发器  
date 是最基本的一种调度，作业任务只会执行一次。它表示特定的时间点触发。它的参数如下：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>run_date (datetime 或 str)</td> 
   <td>作业的运行日期或时间</td> 
  </tr> 
  <tr> 
   <td>timezone (datetime.tzinfo 或 str)</td> 
   <td>指定时区</td> 
  </tr> 
 </tbody> 
</table>

date 触发器使用示例如下：

from datetime import datetime
    from datetime import date
    from apscheduler.schedulers.background import BackgroundScheduler
    
    def job_func(text):
        print(text)
    
    scheduler = BackgroundScheduler()
    # 在 2017-12-13 时刻运行一次 job_func 方法
    scheduler .add_job(job_func, 'date', run_date=date(2017, 12, 13), args=['text'])
    # 在 2017-12-13 14:00:00 时刻运行一次 job_func 方法
    scheduler .add_job(job_func, 'date', run_date=datetime(2017, 12, 13, 14, 0, 0), args=['text'])
    # 在 2017-12-13 14:00:01 时刻运行一次 job_func 方法
    scheduler .add_job(job_func, 'date', run_date='2017-12-13 14:00:01', args=['text'])
    
    scheduler.start()

2）interval 触发器  
固定时间间隔触发。interval 间隔调度，参数如下：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>weeks (int)</td> 
   <td>间隔几周</td> 
  </tr> 
  <tr> 
   <td>days (int)</td> 
   <td>间隔几天</td> 
  </tr> 
  <tr> 
   <td>hours (int)</td> 
   <td>间隔几小时</td> 
  </tr> 
  <tr> 
   <td>minutes (int)</td> 
   <td>间隔几分钟</td> 
  </tr> 
  <tr> 
   <td>seconds (int)</td> 
   <td>间隔多少秒</td> 
  </tr> 
  <tr> 
   <td>start_date (datetime 或 str)</td> 
   <td>开始日期</td> 
  </tr> 
  <tr> 
   <td>end_date (datetime 或 str)</td> 
   <td>结束日期</td> 
  </tr> 
  <tr> 
   <td>timezone (datetime.tzinfo 或str)</td> 
   <td>时区</td> 
  </tr> 
 </tbody> 
</table>

interval 触发器使用示例如下：

import datetime
    from apscheduler.schedulers.background import BackgroundScheduler
    
    def job_func(text):
        print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
    
    scheduler = BackgroundScheduler()
    # 每隔两分钟执行一次 job_func 方法
    scheduler .add_job(job_func, 'interval', minutes=2)
    # 在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之间, 每隔两分钟执行一次 job_func 方法
    scheduler .add_job(job_func, 'interval', minutes=2, start_date='2017-12-13 14:00:01' , end_date='2017-12-13 14:00:10')
    
    scheduler.start()

3）cron 触发器  
在特定时间周期性地触发，和Linux crontab格式兼容。它是功能最强大的触发器。  
我们先了解 cron 参数：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>year (int 或 str)</td> 
   <td>年，4位数字</td> 
  </tr> 
  <tr> 
   <td>month (int 或 str)</td> 
   <td>月 (范围1-12)</td> 
  </tr> 
  <tr> 
   <td>day (int 或 str)</td> 
   <td>日 (范围1-31</td> 
  </tr> 
  <tr> 
   <td>week (int 或 str)</td> 
   <td>周 (范围1-53)</td> 
  </tr> 
  <tr> 
   <td>day_of_week (int 或 str)</td> 
   <td>周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)</td> 
  </tr> 
  <tr> 
   <td>hour (int 或 str)</td> 
   <td>时 (范围0-23)</td> 
  </tr> 
  <tr> 
   <td>minute (int 或 str)</td> 
   <td>分 (范围0-59)</td> 
  </tr> 
  <tr> 
   <td>second (int 或 str)</td> 
   <td>秒 (范围0-59)</td> 
  </tr> 
  <tr> 
   <td>start_date (datetime 或 str)</td> 
   <td>最早开始日期(包含)</td> 
  </tr> 
  <tr> 
   <td>end_date (datetime 或 str)</td> 
   <td>最晚结束时间(包含)</td> 
  </tr> 
  <tr> 
   <td>timezone (datetime.tzinfo 或str)</td> 
   <td>指定时区</td> 
  </tr> 
 </tbody> 
</table>

这些参数是支持算数表达式，取值格式有如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlaGFwcGluZXNz_size_16_color_FFFFFF_t_70][]

cron 触发器使用示例如下：

import datetime
    from apscheduler.schedulers.background import BackgroundScheduler
    
    def job_func(text):
        print("当前时间：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
    
    scheduler = BackgroundScheduler()
    # 在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 job_func 任务
    scheduler .add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')
    
    scheduler.start()

## 4.3 作业存储(job store) ##

该组件是对调度任务的管理。  
1）添加 job  
有两种添加方法，其中一种上述代码用到的 `add_job()`， 另一种则是`scheduled_job()`修饰器来修饰函数。

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

第二种添加任务方式的例子：

import datetime
    from apscheduler.schedulers.background import BackgroundScheduler
    
    @scheduler.scheduled_job(job_func, 'interval', minutes=2)
    def job_func(text):
        print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
    
    scheduler = BackgroundScheduler()
    scheduler.start()

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

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one')
    scheduler.remove_job(job_one)
    
    job = add_job(job_func, 'interval', minutes=2, id='job_one')
    job.remvoe()

3）获取 job 列表  
通过 `scheduler.get_jobs()` 方法能够获取当前调度器中的所有 job 的列表

1.  修改 job  
    如果你因计划改变要对 job 进行修改，可以使用`Job.modify()` 或者 `modify_job()`方法来修改 job 的属性。但是值得注意的是，job 的 id 是无法被修改的。

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one')
    scheduler.start()
    # 将触发时间间隔修改成 5分钟
    scheduler.modify_job('job_one', minutes=5)
    
    job = scheduler.add_job(job_func, 'interval', minutes=2)
    # 将触发时间间隔修改成 5分钟
    job.modify(minutes=5)

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

scheduler.shutdown()
    scheduler.shutdown(wait=false)

## 4.4 执行器(executor) ##

执行器顾名思义是执行调度任务的模块。最常用的 executor 有两种：`ProcessPoolExecutor` 和 `ThreadPoolExecutor`

下面是显式设置 job store(使用mongo存储)和 executor 的代码的示例。  
注：本代码来源于网络

from pymongo import MongoClient
    from apscheduler.schedulers.blocking import BlockingScheduler
    from apscheduler.jobstores.mongodb import MongoDBJobStore
    from apscheduler.jobstores.memory import MemoryJobStore
    from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
     
     
    def my_job():
        print 'hello world'
    host = '127.0.0.1'
    port = 27017
    client = MongoClient(host, port)
     
    jobstores = {
        'mongo': MongoDBJobStore(collection='job', database='test', client=client),
        'default': MemoryJobStore()
    }
    executors = {
        'default': ThreadPoolExecutor(10),
        'processpool': ProcessPoolExecutor(3)
    }
    job_defaults = {
        'coalesce': False,
        'max_instances': 3
    }
    scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
    scheduler.add_job(my_job, 'interval', seconds=5)
     
    try:
        scheduler.start()
    except SystemExit:
        client.close()

原文链接：[https://www.jianshu.com/p/d04bd534b219][https_www.jianshu.com_p_d04bd534b219]   [https://www.jianshu.com/p/ad2c42245906][https_www.jianshu.com_p_ad2c42245906]  
来源：简书  
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

写博客是为了记住自己容易忘记的东西，另外也是对自己工作的总结，希望尽自己的努力，做到更好，大家一起努力进步！  
  
如果有什么问题，欢迎大家一起探讨，代码如有问题，欢迎各位大神指正！  
  
给自己的梦想添加一双翅膀，让它可以在天空中自由自在的飞翔！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlaGFwcGluZXNz_size_16_color_FFFFFF_t_70]: /images/20210923/aa7a7d109ac345739267b5332d162b62.png
[https_www.jianshu.com_p_d04bd534b219]: https://www.jianshu.com/p/d04bd534b219
[https_www.jianshu.com_p_ad2c42245906]: https://www.jianshu.com/p/ad2c42245906