python_APScheduler任务调度

蔚落 2022-01-22 11:47 312阅读 0赞

## Installing APScheduler ##

The preferred installation method is by using [pip][]:

pip install apscheduler

### Code examples ###

The source distribution contains the `examples` directory where you can find many working examples for using APScheduler in different ways. The examples can also be [browsed online][].

## Basic concepts ##

APScheduler has four kinds of components:

*  triggers
 *  job stores
 *  executors
 *  schedulers

triggers触发器包含调度逻辑。每个作业都有自己的触发器，用于确定下一个作业何时运行。除了初始配置之外，触发器是完全无状态的。

job stores默认的作业存储只是将作业保存在内存中，而其他的则将它们存储在各种数据库中。作业的数据在保存到持久性作业存储时被序列化，在从中加载后被反序列化。作业存储区（默认存储区除外）不将作业数据保存在内存中，而是充当后台保存、加载、更新和搜索作业的中间人。决不能在调度程序之间共享作业存储。

executors执行者是处理作业运行的人。它们通常通过将作业中指定的可调用项提交到线程或进程池来实现这一点。作业完成后，执行器通知调度程序，调度程序随后发出适当的事件。

schedulers调度程序是将其余部分绑定在一起的工具。您的应用程序中通常只有一个调度程序在运行。应用程序开发人员通常不直接处理作业存储、执行器或触发器。相反，调度程序提供适当的接口来处理所有这些问题。配置作业存储和执行器是通过调度程序完成的，添加、修改和删除作业也是如此。

## Choosing the right scheduler, job store(s), executor(s) and trigger(s) ##

您对调度器的选择主要取决于您的编程环境以及您将使用apscheduler的目的。以下是选择计划程序的快速指南：

*  blockingscheduler：当调度程序是进程中唯一运行的东西时使用
 *  BackgroundScheduler：当您不使用下面的任何框架，并且希望该调度程序在应用程序内部的后台运行时使用。
 *  AsyncioScheduler:如果应用程序使用Asyncio模块，则使用
 *  geventscheduler：如果应用程序使用gevent，则使用
 *  TornadoScheduler:在构建Tornado应用程序时使用
 *  TwistedScheduler:在构建扭曲的应用程序时使用
 *  qtscheduler：在构建qt应用程序时使用

To pick the appropriate job store, you need to determine whether you need job persistence or not. If you always recreate your jobs at the start of your application, then you can probably go with the default (MemoryJobStore). But if you need your jobs to persist over scheduler restarts or application crashes, then your choice usually boils down to what tools are used in your programming environment. If, however, you are in the position to choose freely, then SQLAlchemyJobStore on a PostgreSQL backend is the recommended choice due to its strong data integrity protection.

Likewise, the choice of executors is usually made for you if you use one of the frameworks above. Otherwise, the default ThreadPoolExecutor should be good enough for most purposes. If your workload involves CPU intensive operations, you should consider using ProcessPoolExecutor instead to make use of multiple CPU cores. You could even use both at once, adding the process pool executor as a secondary executor.

When you schedule a job, you need to choose a trigger for it. The trigger determines the logic by which the dates/times are calculated when the job will be run. APScheduler comes with three built-in trigger types:

*  [`date`][date]: use when you want to run the job just once at a certain point of time
 *  [`interval`][interval]: use when you want to run the job at fixed intervals of time
 *  [`cron`][cron]: use when you want to run the job periodically at certain time(s) of day

也可以将多个触发器组合成一个触发器，在所有参与触发器约定的时间或任何触发器将触发时触发。有关详细信息，请参阅有关组合触发器的文档。

您可以在各自的API文档页面上找到每个作业存储、执行器和触发器类型的插件名称。

## Configuring the scheduler ##

apscheduler提供了许多不同的方法来配置调度程序。可以使用配置字典，也可以将选项作为关键字参数传入。您还可以先实例化调度程序，然后添加作业和配置调度程序。这样，您就可以为任何环境获得最大的灵活性。

可以在BaseScheduler类的API引用中找到计划程序级配置选项的完整列表。调度器子类还可以有其他选项，这些选项记录在各自的API引用中。各个作业存储和执行器的配置选项也可以在其API参考页上找到。

假设您希望在应用程序中使用默认的作业存储和默认的执行器运行backgroundscheduler：

from apscheduler.schedulers.background import BackgroundScheduler
    
    
    scheduler = BackgroundScheduler()
    
    # Initialize the rest of the application here, or before the scheduler initialization

这将为您提供一个backgroundscheduler，其中memoryjobstore名为“default”，threadpoolexecutor名为“default”，默认最大线程数为10。

现在，假设你想要更多。您希望有两个使用两个执行器的作业存储，还希望调整新作业的默认值并设置不同的时区。以下三个示例完全等效，将为您提供：

a MongoDBJobStore named “mongo”  
an SQLAlchemyJobStore named “default” (using SQLite)  
a ThreadPoolExecutor named “default”, with a worker count of 20  
a ProcessPoolExecutor named “processpool”, with a worker count of 5  
UTC as the scheduler’s timezone  
coalescing turned off for new jobs by default  
a default maximum instance limit of 3 for new jobs

Method 1:

from pytz import utc
    
    from apscheduler.schedulers.background import BackgroundScheduler
    from apscheduler.jobstores.mongodb import MongoDBJobStore
    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
    from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
    
    
    jobstores = {
        'mongo': MongoDBJobStore(),
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
    }
    executors = {
        'default': ThreadPoolExecutor(20),
        'processpool': ProcessPoolExecutor(5)
    }
    job_defaults = {
        'coalesce': False,
        'max_instances': 3
    }
    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

Method 2:

from apscheduler.schedulers.background import BackgroundScheduler
    
    
    # The "apscheduler." prefix is hard coded
    scheduler = BackgroundScheduler({
        'apscheduler.jobstores.mongo': {
             'type': 'mongodb'
        },
        'apscheduler.jobstores.default': {
            'type': 'sqlalchemy',
            'url': 'sqlite:///jobs.sqlite'
        },
        'apscheduler.executors.default': {
            'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
            'max_workers': '20'
        },
        'apscheduler.executors.processpool': {
            'type': 'processpool',
            'max_workers': '5'
        },
        'apscheduler.job_defaults.coalesce': 'false',
        'apscheduler.job_defaults.max_instances': '3',
        'apscheduler.timezone': 'UTC',
    })

Method 3:

from pytz import utc
    
    from apscheduler.schedulers.background import BackgroundScheduler
    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
    from apscheduler.executors.pool import ProcessPoolExecutor
    
    
    jobstores = {
        'mongo': {'type': 'mongodb'},
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
    }
    executors = {
        'default': {'type': 'threadpool', 'max_workers': 20},
        'processpool': ProcessPoolExecutor(max_workers=5)
    }
    job_defaults = {
        'coalesce': False,
        'max_instances': 3
    }
    scheduler = BackgroundScheduler()
    
    # .. do something else here, maybe add jobs etc.
    
    scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

### 启动计划程序 ###

启动调度程序只需在调度程序上调用start（）即可完成。对于blockingscheduler以外的调度程序，此调用将立即返回，您可以继续应用程序的初始化过程，可能会向调度程序添加作业。

启动计划程序后，您不能再更改其设置。

## 添加任务 ##

向调度程序添加作业有两种方法：

1.  by calling [`add_job()`][add_job]
2.  by decorating a function with [`scheduled_job()`][scheduled_job]

第一种方法是最常见的方法。第二种方法主要是方便地声明在应用程序运行时不会更改的作业。add\_job（）方法返回一个apscheduler.job.job实例，您以后可以使用该实例修改或删除作业。

您可以随时在调度程序上调度作业。如果添加作业时调度程序尚未运行，则将暂时调度作业，并且仅在调度程序启动时计算其第一次运行时间。

需要注意的是，如果您使用序列化作业的执行器或作业存储，它将为您的作业添加一些要求：

*  The target callable must be globally accessible
 *  Any arguments to the callable must be serializable

在内置的作业存储中，只有memoryjobstore不序列化作业。在内置执行器中，只有ProcessPoolexecutor将序列化作业。

--------------------

**Important**

如果在应用程序初始化期间在持久性作业存储中计划作业，则必须为该作业定义一个显式ID，并使用replace\_existing=true，否则每次应用程序重新启动时，都将获得该作业的新副本！

Tip

要立即运行作业，请在添加作业时省略触发器参数。

--------------------

## Removing jobs ##

When you remove a job from the scheduler, it is removed from its associated job store and will not be executed anymore. There are two ways to make this happen:

1.  by calling [`remove_job()`][remove_job] with the job’s ID and job store alias
2.  by calling [`remove()`][remove] on the Job instance you got from [`add_job()`][add_job]

后一种方法可能更方便，但它要求将添加作业时收到的作业实例存储在某个位置。对于通过scheduled\_job（）计划的作业，第一种方法是唯一的方法。

如果作业的计划结束（即其触发器不会产生任何进一步的运行时间），则会自动删除该计划。

Example:

job = scheduler.add_job(myfunc, 'interval', minutes=2)
    job.remove()

Same, using an explicit job ID:

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
    scheduler.remove_job('my_job_id')

## Pausing and resuming jobs ##

您可以通过作业实例或调度程序本身轻松地暂停和恢复作业。当一个作业暂停时，它的下一个运行时间将被清除，在作业恢复之前，不会再为它计算任何运行时间。要暂停作业，请使用以下任一方法：

*  [`apscheduler.job.Job.pause()`][apscheduler.job.Job.pause]
 *  [`apscheduler.schedulers.base.BaseScheduler.pause_job()`][apscheduler.schedulers.base.BaseScheduler.pause_job]

To resume:

*  [`apscheduler.job.Job.resume()`][apscheduler.job.Job.resume]
 *  [`apscheduler.schedulers.base.BaseScheduler.resume_job()`][apscheduler.schedulers.base.BaseScheduler.resume_job]

要获取计划作业的机器可处理列表，可以使用get\_jobs（）方法。它将返回作业实例列表。如果您只对特定作业存储区中包含的作业感兴趣，那么请提供一个作业存储别名作为第二个参数。

为了方便起见，您可以使用print\_jobs（）方法，该方法将打印作业及其触发器和下一次运行时间的格式化列表。

## Modifying jobs ##

You can modify any job attributes by calling either [`apscheduler.job.Job.modify()`][apscheduler.job.Job.modify] or [`modify_job()`][modify_job]. You can modify any Job attributes except for `id`.

Example:

job.modify(max_instances=6, name='Alternate name')

如果要重新安排作业，也就是更改其触发器，可以使用eitherapscheduler.job.job.resschedule（）或resschedule\_job（）。这些方法为作业构造一个新触发器，并基于新触发器重新计算其下一个运行时间。

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

## 限制同时执行作业的实例数 ##

默认情况下，只允许同时运行每个作业的一个实例。这意味着，如果作业即将运行，但上一次运行尚未完成，则最近一次运行被视为失火。通过在添加作业时使用max\_instances关键字参数，可以设置计划程序允许并发运行的特定作业的最大实例数。

## Missed job executions and coalescing ##

有时计划程序可能无法在计划运行时执行计划作业。最常见的情况是在持久性作业存储中调度作业，并且调度程序在作业应该执行之后关闭并重新启动。当这种情况发生时，工作被认为是“缺火”。然后，调度程序将对照作业的缺火宽限期选项（可以在每个作业或调度程序中全局设置）检查每个错过的执行时间，以查看是否仍应触发执行。这会导致作业连续执行几次。

如果这种行为不适合您的特定用例，那么可以使用合并将所有这些错过的执行回滚到一个用例中。换句话说，如果对作业启用了合并，并且调度程序看到该作业的一个或多个排队执行，那么它将只触发一次。对于“旁路”运行，不会发送缺火事件。

# 简单应用： #

import time
    from apscheduler.schedulers.blocking import BlockingScheduler
     
    def my_job():
        print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
     
    sched = BlockingScheduler()
    sched.add_job(my_job, 'interval', seconds=5)
    sched.start()

## 移除作业 ##

job = scheduler.add_job(myfunc, 'interval', minutes=2)
    job.remove()
    #如果有多个任务序列的话可以给每个任务设置ID号，可以根据ID号选择清除对象，且remove放到start前才有效
    sched.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
    sched.remove_job('my_job_id')

## 暂停和恢复作业 ##

暂停作业:

apsched.job.Job.pause()
    apsched.schedulers.base.BaseScheduler.pause_job()

恢复作业:

apsched.job.Job.resume()
    apsched.schedulers.base.BaseScheduler.resume_job()

## 获得job列表 ##

获得调度作业的列表，可以使用`get_jobs()`来完成，它会返回所有的job实例。或者使用`print_jobs()`来输出所有格式化的作业列表。也可以利用get\_job(任务ID)获取指定任务的作业列表

job = sched.add_job(my_job, 'interval', seconds=2 ,id='123')
    print sched.get_job(job_id='123')
    print sched.get_jobs()

## 关闭调度器 ##

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

sched.shutdown()
    sched.shutdown(wait=False)

** 作业运行的控制（trigger）**

add\_job的第二个参数是trigger，它管理着作业的调度方式。它可以为date, interval或者cron。对于不同的trigger，对应的参数也相同。

## (1). 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) -（表示时区取值）

### 参数的取值格式 ###

![940167-20170322175636377-973994665.png][]

### 例子： ###

#表示2019年6月10日10时40分00秒执行该程序
    sched.add_job(my_job, 'cron', year=2019,month = 6,day = 10,hour = 10,minute = 40,second = 0)
    
     
    #表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序
    sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
     
    #表示从星期一到星期五5:30（AM）直到2014-05-30 00:00:00
    sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2014-05-30')
     
    #表示每5秒执行该程序一次，相当于interval 间隔调度中seconds = 5
    sched.add_job(my_job, 'cron',second = '*/5')

## interval 间隔调度（每隔多久执行） ##

weeks (int) – number of weeks to wait
    days (int) – number of days to wait
    hours (int) – number of hours to wait
    minutes (int) – number of minutes to wait
    seconds (int) – number of seconds to wait
    start_date (datetime|str) – starting point for the interval calculation
    end_date (datetime|str) – latest possible date/time to trigger on
    timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

### 例子： ###

#表示每隔3天17时19分07秒执行一次任务
    sched.add_job(my_job, 'interval',days  = 03,hours = 17,minutes = 19,seconds = 07)

## 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

例子

# The job will be executed on November 6th, 2009
    sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
    # The job will be executed on November 6th, 2009 at 16:30:05
    sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])

[pip]: http://pypi.python.org/pypi/pip/
[browsed online]: https://github.com/agronholm/apscheduler/tree/master/examples/?at=master
[date]: https://apscheduler.readthedocs.io/en/latest/modules/triggers/date.html#module-apscheduler.triggers.date
[interval]: https://apscheduler.readthedocs.io/en/latest/modules/triggers/interval.html#module-apscheduler.triggers.interval
[cron]: https://apscheduler.readthedocs.io/en/latest/modules/triggers/cron.html#module-apscheduler.triggers.cron
[add_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.add_job
[scheduled_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.scheduled_job
[remove_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.remove_job
[remove]: https://apscheduler.readthedocs.io/en/latest/modules/job.html#apscheduler.job.Job.remove
[apscheduler.job.Job.pause]: https://apscheduler.readthedocs.io/en/latest/modules/job.html#apscheduler.job.Job.pause
[apscheduler.schedulers.base.BaseScheduler.pause_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.pause_job
[apscheduler.job.Job.resume]: https://apscheduler.readthedocs.io/en/latest/modules/job.html#apscheduler.job.Job.resume
[apscheduler.schedulers.base.BaseScheduler.resume_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.resume_job
[apscheduler.job.Job.modify]: https://apscheduler.readthedocs.io/en/latest/modules/job.html#apscheduler.job.Job.modify
[modify_job]: https://apscheduler.readthedocs.io/en/latest/modules/schedulers/base.html#apscheduler.schedulers.base.BaseScheduler.modify_job
[940167-20170322175636377-973994665.png]: /images/20220122/fec0859176e544bcbfc70dbe32c04f1f.png