当前位置: 首页 > biancheng >正文

APScheduler 定时任务模块实现

目录

定时任务简介

安装

一、调度器(schedulers)

二、任务存储器(job stores)

三、触发器

四、执行器

五、定时任务 调度配置

六、任务操作

七、异常监听与日志

八、示例代码

九、管理定时任务,实现对任务的增、删、改、查操作


定时任务简介

    APScheduler使用起来十分方便。提供了基于日期、固定时间间隔以及 crontab类型的任务。还可以在程序运行过程中动态的新增任务和删除任务。在任务运行过程中,还可以把任务存储起来,下次启动运行依然保留之前的状态。另外最重要的一个特点是,因为他是基于 Python语言的库,所以是可以跨平台的,一段代码,处处运行。

 

安装

pip install apscheduler

一、调度器(schedulers)

    任务调度器是属于整个调度的总指挥官。他会合理安排作业存储器、执行器、触发器进行工作,并进行添加和删除任务等。调度器通常是只有一个的。开发人员很少直接操作触发器、存储器、执行器等。因为这些都由调度器自动来实现了。

  1. BlockingScheduler :适用于调度程序是进程中唯一运行的进程,调用 start 函数会阻塞当前线程,不能立即返回。
  2. BackgroundScheduler :适用于调度程序在应用程序的后台运行,调用 start 后主线程不会阻塞。
  3. AsyncIOScheduler :适用于使用了 asyncio 模块的应用程序。
  4. GeventScheduler :适用于使用 gevent 模块的应用程序。
  5. TwistedScheduler :适用于构建 Twisted 的应用程序。
  6. QtScheduler :适用于构建 Qt 的应用程序。

二、任务存储器(job stores)

    任务存储器是可以存储任务的地方,默认情况下任务保存在内存,也可将任务保存在各种数据库中。任务存储进去后,会进行序列化,然后也可以反序列化提取出来,继续执行。

  1. MemoryJobStore :没有序列化,任务存储在内存中,增删改查都是在内存中完成。
  2. SQLAlchemyJobStore :使用 SQLAlchemy 这个 ORM 框架作为存储方式。
  3. MongoDBJobStore :使用 mongodb 作为存储器。
  4. RedisJobStore :使用 redis 作为存储器。

    任务存储器的选择有两种。一是内存,也是默认的配置。二是数据库。使用内存的方式是简单高效,但是不好的是,一旦程序出现问题,重新运行的话,会把之前已经执行了的任务重新执行一遍。数据库则可以在程序崩溃后,重新运行可以从之前中断的地方恢复正常运行。

三、触发器

    触发器有3种,第一种是date,第二种是 interval,第三种是 crontab。 interval可以具体指定多少时间间隔执行一次。 crontab可以指定执行的日期策略。

1、date触发器:

    在某个日期时间只触发一次事件。

    示例:sched . add_job ( my_job , 'date' , run_date = date ( 2020 , 5 , 22 ), args =[ 'text' ])

2、interval触发器

    想要在固定的时间间隔触发事件。 interval的触发器可以设置以下的触发参数: 

  示例:scheduler . add_job ( tick , "cron" , hour = 11 , minute = 24 )

3、crontab触发器

    在某个确切的时间周期性的触发事件,定时任务cron表达式

 示例:scheduler . add_job ( tick , "cron" , day = "4th sun" , hour = 20 , minute = 1 )

四、执行器

  1. ThreadPoolExecutor :线程池执行器。
  2. ProcessPoolExecutor :进程池执行器。
  3. GeventExecutor : Gevent 程序执行器。
  4. TornadoExecutor : Tornado 程序执行器。
  5. TwistedExecutor : Twisted 程序执行器。
  6. AsyncIOExecutor : asyncio 程序执行器。

五、定时任务 调度配置

    1、执行器:配置 default 执行器为 ThreadPoolExecutor ,并且设置最多的线程数是x个。

    2、存储器:本系统采用默认存储在运行内存中管理

    3、任务配置:

  • 3.1 coalesce = True,设置这个目的是,比如由于某个原因导致某个任务积攒了很多次没有执行(比如有一个任务是1分钟跑一次,但是系统原因断了5分钟),如果 coalesce = True ,那么下次恢复运行的时候会只执行一次,而如果设置 coalesce = False ,那么就不会合并会5次全部执行。
  • 3.2 max_instances = 5,同一个任务同一时间最多只能有5个实例在运行。比如一个耗时10分钟的job,被指定每分钟运行1次,如果我 max_instance 值5,那么在第 6 ~ 10 分钟上,新的运行实例不会被执行,因为已经有5个实例在跑了。

六、任务操作

 1. 添加任务:
 scheduler.add_job(job_obj,args,id,trigger,**trigger_kwargs)。

 2. 删除任务:
 scheduler.remove_job(job_id,jobstore=None)。

 3. 暂停任务:
 scheduler.pause_job(job_id,jobstore=None)。

 4. 恢复任务:
 scheduler.resume_job(job_id,jobstore=None)。

 5. 修改某个任务属性信息:
 scheduler.modify_job(job_id,jobstore=None,**changes)。

 6. 修改单个作业的触发器并更新下次运行时间:
 scheduler.reschedule_job(job_id,jobstore=None,trigger=None,**trigger_args)

 7. 输出作业信息:
 scheduler.print_jobs(jobstore=None,out=sys.stdout)

七、异常监听与日志

    当我们的任务抛出异常后,我们可以监听到,然后把错误信息进行记录。

八、示例代码

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR, EVENT_JOB_MISSED
from apscheduler.triggers.cron import CronTrigger
import logging
logger = logging.getLogger('job')
logging.basicConfig(level=logging.INFO,
    format = '%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
    datefmt = '%Y-%m-%d %H:%M:%S',
    filename = 'log/TaskChedulerLog.txt',
    filemode = 'a',
    # encoding = 'utf-8', # Python 低版本不支持该配置
)


# 重写Cron表达式 提供6个参数进行处理,不支持?
class my_CronTrigger(CronTrigger):
    @classmethod
    def my_from_crontab(cls, expr, timezone="Asia/Shanghai"):
        values = expr.split()
        if len(values) != 6:
            raise ValueError('Wrong number of fields; got {}, expected 7'.format(len(values)))
        return cls(second=values[0], minute=values[1], hour=values[2], day=values[3], month=values[4],day_of_week=values[5], timezone=timezone)


# 定时任务 任务监听器
def my_listener(event):
    job = scheduler.get_job(event.job_id)
    if event.exception:
        logger.info("#### job_error Task error!!!!!!")
        logger.info(event.exception)
        logger.info("#### job_id=%s|obname=%s|jobtrigger=%s|jobtime=%s|retval=%s", event.job_id, job.name, job.trigger,event.scheduled_run_time, event.retval)

    else:
        logger.info('#### jon_normal Task Running...')
        logger.info("#### job_id=%s|jobname=%s|jobtrigger=%s|jobtime=%s|retval=%s", event.job_id,job.name, job.trigger, event.scheduled_run_time, event.retval)


# 定义定时 调度配置
executors = {
    'default': ThreadPoolExecutor(20)
}
job_defaults = {
    'coalesce': True,
 'max_instances': 1
}

# 定时任务 初始化
scheduler = BackgroundScheduler(executors = executors, job_defaults=job_defaults,timezone="Asia/Shanghai")
scheduler.add_listener(my_listener, EVENT_JOB_ERROR | EVENT_JOB_MISSED | EVENT_JOB_EXECUTED)
scheduler._logger = logger

# Cron 6位:* * * * * * 秒、分、时、天、月、周
# project_id,project_name,branch,path,run_env = "1","国内酒店","testSute","/test","dev-test"
# interval_task 为调用Jenkins的函数方法,或也可为调试定时任务执行情况


# scheduler.add_job(interval_task ,my_CronTrigger.my_from_crontab("*/2 * * * * *"),args=[project_name,branch,path,run_env],id=project_id)
# scheduler.add_job(getProJectPlanInfo ,my_CronTrigger.my_from_crontab("*/2 * * * * *"))

# 定义每间隔15分钟更新一次配置,getProJectPlanInfo函数方法为查询数据库定时任务配置筛选启用的定时任务,并对定时任务进行 增、删、改、查操作
scheduler.add_job(getProJectPlanInfo ,my_CronTrigger.my_from_crontab("0 */15 * * * *"))

# 显示timing 任务列表
# scheduler.print_jobs(jobstore=None,out=sys.stdout)

scheduler.start()

九、管理定时任务,实现对任务的增、删、改、查操作

    1、定时任务信息dict,用户存储现有定时任务的信息如:项目id、项目定时计划cron表达式、项目运行目录、运行环境等

    2、定时任务信息dict-keyList,现有运行定时任务的keyList,方便管理运行中的job_id

    3、每15分钟获取一次已设置定时任务的project_id List,获取当前可运行的定时任务。通过获取步骤1的信息“查”询判断定时任务 增、删、改操作

    4、对比 步骤2、步骤3 的数据对,完成对定时任务信息dict、定时任务信息dict-keyList完成数据更新,同时移除已关闭的定时任务

相关文章:

  • 牛客练习赛#84 F 莫比乌斯反演+杜教筛+技巧+斐波那契数列和gcd的结论+矩阵快速幂
  • ZZNUOJ_用C语言编写程序实现1342:支配值数目(附完整源码)
  • java毕业设计后勤管理系统餐饮评价监督系统(附源码、数据库)
  • 前端基础学习笔记
  • 【TS】联合类型--类型断言--类型推断
  • 谈笑风声的秘密
  • QT影城网上售票系统
  • NetCDF数据在ArcMap中的使用
  • 打怪升级(考验思路)
  • 持续精进,改变自己