Python中怎么使用apscheduler定时执行任务

今天小编给大家分享一下Python中怎么使用apscheduler定时执行任务的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。

apscheduler 的使用

  我们项目中总是避免不了要使用一些定时任务，比如说最近的项目，用户点击报名考试以后需要在考试日期临近的时候推送小程序消息提醒到客户微信上，翻了翻 fastapi 中的实现，虽然方法和包也不少，但是要不就是太重了（比如需要再开服务，还要依赖 redis，都不好用），虽然也可以使用 time 模块的 time.sleep()机上 fastapi 的后台任务变相实现，但是相对简单的功能还行，复杂点的代码起来就麻烦了，所以还是专人专事找个负责这个额的包吧。找来找去发现 APScheduler 就挺适合，代码简单，实现效果也很好，这里做个记录！

安装

pip install apscheduler

主要组成部分

概念性东西，混个脸熟，代码比这些定义好理解。

触发器(trigger)包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的。说人话就是你指定那种方式触发当前的任务。

作业存储(job store)存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。

Jobstore在scheduler中初始化，另外也可通过scheduler的add_jobstore动态添加Jobstore。每个jobstore
都会绑定一个alias，scheduler在Add Job时，根据指定的jobstore在scheduler中找到相应的jobstore，并
将job添加到jobstore中。

Jobstore主要是通过pickle库的loads和dumps【实现核心是通过python的__getstate__和__setstate__重写
实现】，每次变更时将Job动态保存到存储中，使用时再动态的加载出来，作为存储的可以是redis，也可以
是数据库【通过sqlarchemy这个库集成多种数据库】，也可以是mongodb等
目前APScheduler支持的Jobstore：

MemoryJobStore
MongoDBJobStore
RedisJobStore
RethinkDBJobStore
SQLAlchemyJobStore
ZooKeeperJobStore

执行器(executor)处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。

- 说人话就是添加任务时候用它来包装的，executor的种类会根据不同的调度来选择，如果选择AsyncIO作为调度的库，那么选择AsyncIOExecutor，如果选择tornado作为调度的库，选择TornadoExecutor，如果选择启动进程作为调度，选择ThreadPoolExecutor或者ProcessPoolExecutor都可以Executor的选择需要根据实际的scheduler来选择不同的执行器
目前APScheduler支持的Executor：
AsyncIOExecutor
GeventExecutor
ThreadPoolExecutor
ProcessPoolExecutor
TornadoExecutor
TwistedExecutor

调度器(scheduler)是其他的组成部分。你通常在应用只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业.

Scheduler是APScheduler的核心，所有相关组件通过其定义。scheduler启动之后，将开始按照配置的任务进行调度。
除了依据所有定义Job的trigger生成的将要调度时间唤醒调度之外。当发生Job信息变更时也会触发调度。

scheduler可根据自身的需求选择不同的组件，如果是使用AsyncIO则选择AsyncIOScheduler，使用tornado则
选择TornadoScheduler。
目前APScheduler支持的Scheduler：

AsyncIOScheduler
BackgroundScheduler
BlockingScheduler
GeventScheduler
QtScheduler
TornadoScheduler
TwistedScheduler

简单应用

import timefrom 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()

完整代码

# trigeers 触发器# job stores job 存储# executors 执行器# schedulers 调度器 from pytz import utcfrom sqlalchemy import func from apscheduler.schedulers.background import BackgroundScheduler,AsyncIOSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ProcessPoolExecutor  jobstores = {    # 可以配置多个存储    #'mongo': {'type': 'mongodb'},     'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')  # SQLAlchemyJobStore指定存储链接}executors = {    'default': {'type': 'threadpool', 'max_workers': 20},     # 最大工作线程数20    'processpool': ProcessPoolExecutor(max_workers=5)         # 最大工作进程数为5}job_defaults = {    'coalesce': False,   # 关闭新job的合并，当job延误或者异常原因未执行时    'max_instances': 3   # 并发运行新job默认最大实例多少}scheduler = BackgroundScheduler() # .. do something else here, maybe add jobs etc. scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc) # utc作为调度程序的时区  import osimport time def print_time(name):    print(f'{name} - {time.ctime()}') def add_job(job_id, func, args, seconds):    """添加job"""    print(f"添加间隔执行任务job - {job_id}")    scheduler.add_job(id=job_id, func=func, args=args, trigger='interval', seconds=seconds) def add_coun_job(job_id, func, args, start_time):    """添加job"""    print(f"添加一次执行任务job - {job_id}")    scheduler.add_job(id=job_id, func=func, args=args, trigger='date',timezone='Asia/Shanghai', run_date=start_time)    # scheduler.add_job(func=print_time, trigger='date',timezone='Asia/Shanghai', run_date=datetime(2022, 2, 19, 17, 57, 0).astimezone(), args=['text2']) def remove_job(job_id):    """移除job"""    scheduler.remove_job(job_id)    print(f"移除job - {job_id}") def pause_job(job_id):    """停止job"""    scheduler.pause_job(job_id)    print(f"停止job - {job_id}") def resume_job(job_id):    """恢复job"""    scheduler.resume_job(job_id)    print(f"恢复job - {job_id}") def get_jobs():    """获取所有job信息,包括已停止的"""    res = scheduler.get_jobs()    print(f"所有job - {res}") def print_jobs():    print(f"详细job信息")    scheduler.print_jobs() def start():    """启动调度器"""    scheduler.start() def shutdown():    """关闭调度器"""    scheduler.shutdown() if __name__ == '__main__':    scheduler = BackgroundScheduler()    # start()    # print('Press Ctrl+{0} to exit \n'.format('Break' if os.name == 'nt' else 'C'))    # add_job('job_A', func=print_time, args=("A", ), seconds=1)    # add_job('job_B', func=print_time, args=("B", ), seconds=2)    # time.sleep(6)    # pause_job('job_A') # 停止a    # get_jobs()   #得到所有job    # time.sleep(6)    # print_jobs()    # resume_job('job_A')    # time.sleep(6)    # remove_job('job_A')    # time.sleep(6)    from datetime import datetime    import pytz    start()        date_temp = datetime(2022, 2, 19, 17, 30, 5)    # scheduler.add_job(print_time, 'date', run_date=datetime.now(pytz.timezone('America/Manaus')), args=['text'])    # scheduler.add_job(print_time, 'date',timezone='Asia/Shanghai', run_date=datetime(2022, 2, 19, 17, 57, 0).astimezone(), args=['text2'])    add_coun_job(job_id="job_C",func=print_time,args=('一次性执行任务',),start_time=datetime(2022, 2, 19, 18, 4, 0).astimezone())    time.sleep(130)    try:        shutdown()    except RuntimeError:        pass

以上就是“Python中怎么使用apscheduler定时执行任务”这篇文章的所有内容，感谢各位的阅读！相信大家阅读完这篇文章都有很大的收获，小编每天都会为大家更新不同的知识，如果还想学习更多的知识，请关注编程网行业资讯频道。