apscheduler-定时任务笔记

  apscheduler定时任务笔记  文章目录1、创建一个简单的定时任务代码 说明 关于失联党员情况说明岗位说明总经理岗位说明书会计岗位说明书行政主管岗位说明书 ：1.1、实例化一个调度器1.2、指定一个触发器(triggers)1.2.1使用触发器interval后，可跟的参数有：1.2.2使用触发器date后，可跟的参数有:1.2.2使用触发器cron后，可跟的参数有:2、调度器、作业存储器、执行器详细说明2.1开始任务前，一般先对调度器进行配置。2.2作业存储器（jobstores）说明2.3执行器配置2.4、启动和关闭调度器3、添加工作定时任务3.1添加任务的方法有两种：3.2删除任务4、事件监听1、创建一个简单的定时任务建一个每隔多少时间运行一次的任务进行说明fromapscheduler.schedulers.blockingimportBlockingSchedulerfromdatetimeimportdatetimedefmy_clock():print(f"调用时间:{datetime.now()}")sched=BlockingScheduler()#实例化一个调度器#每隔2秒调用一次my_clock()函数sched.add_job(my_clock,trigger='interval',seconds=2)print(f'开始时间：{datetime.now()}')sched.start()ps：函数调用是在start()开始2秒后才开始第一次调用的。开始时间：2021-03-2322:48:27.174595调用时间:2021-03-2322:48:29.186497#两秒调用一次，即2秒后才开始第一次调用函数。调用时间:2021-03-2322:48:31.190019……代码说明：1.1、实例化一个调度器sched=BlockingScheduler()BlockingScheduler()：调度器是阻塞的，即同个线程内scheduler.start()之后的代码都不会被运行。还有其它6个调度器，分别为：1、BackgroundScheduler: 会开启一个守护线程，即不阻塞主线程的运行，当主线程运行结束后scheduler也自动停止。默认最大线程数为10个。2、AsyncIOScheduler: 在协程asyncio中应用。3、GeventScheduler: 在gevent框架中使用。4、TornadoScheduler: 在Tornado框架中使用。5、TwistedScheduler: 在Twisted框架的应用。6、QtScheduler: 适合使用QT的情况。注：如果用BackgroundScheduler调度器来执行sqlalchemy的数据库操作，可能会报线程错误。比如：sqlalchemy.exc.ProgrammingError:(sqlite3.ProgrammingError)SQLiteobjectscreatedinathreadcanonlybeusedinthatsamethread.……此时得在create_engine创建引擎时加入参数connect_args={‘check_same_thread’:False}如：engine=create_engine('sqlite+pysqlite:///..\\name.db',future=True,connect_args={'check_same_thread':False})1.2、指定一个触发器(triggers)sched.add_job(my_clock,trigger='interval',seconds=2)由trigger参数指定触发器，来指定调用函数的方式。可选择的有：1、interval：以固定的时间间隔执行。2、date：在特定的时间日期执行。3、cron：指定在某个时间点执行或循环执行。1.2.1使用触发器interval后，可跟的参数有：weeks(int)–间隔的周数days(int)–间隔的天数hours(int)–间隔的小时minutes(int)–间隔的分钟seconds(int)–间隔的秒start_date(datetime|str)–开始时间，如果start_date为空，则默认是datetime.now()+interval作为起始时间。end_date(datetime|str)–结束时间timezone(datetime.tzinfo|str)–时区jitter(int|None)–波动参数，给每次触发添加一个范围内随机浮动秒数。一般适用于多服务器，避免同时运行造成服务拥堵例：fromdatetimeimportdatetime#每两小时执行一次sched.add_job(job_function,'interval',hours=2)#从2021年3月24日22:19:00到2021年12月31日23:59:59的时间内，每两小时执行一次。sched.add_job(job_function,'interval',hours=2,start_date='2021-3-2422:19:00',end_date='2021-12-3123:59:59')#next_run_time设置第一次调用的时间。#下面代码如果不设置next_run_time，则会在start()后两小时才开始第一次调用。#设置next_run_time=datetime.datetime.now()后，则马上开始第一次调用。sched.add_job(main,'interval',hours=2,next_run_time=datetime.datetime.now())1.2.2使用触发器date后，可跟的参数有:run_date(datetime|str)–任务开始的时间，可以是datetime类型，也可以是字符串，如果run_date为空，则默认取当前时间。timezone(datetime.tzinfo|str)–时区。例：#在2021年3月24日22:42:30调用函数my_clock。sched.add_job(my_clock,trigger='date',run_date=datetime(2021,3,24,22,42,30))sched.add_job(my_clock,trigger='date',run_date="2021-3-2422:42:30")1.2.2使用触发器cron后，可跟的参数有:year(int|str)–4digityear-（表示四位数的年份，如2008年）month(int|str)–month(1-12)-（表示取值范围为1-12月）day(int|str)–dayofthe(1-31)-（表示取值范围为1-31日）week(int|str)–ISOweek(1-53)-（格里历2006年12月31日可以写成2006年-W52-7（扩展形式）或2006W527（紧凑形式））day_of_week(int|str)–numberornameofweekday(0-6ormon,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)–earliestpossibledate/timetotriggeron(inclusive)-（表示开始时间）end_date(datetime|str)–latestpossibledate/timetotriggeron(inclusive)-（表示结束时间）timezone(datetime.tzinfo|str)–timezonetouseforthedate/timecalculations(defaultstoschedulertimezone)-（表示时区取值）参数格式：序号格式可用参数说明1*所有表示每个单位的该参数。如hour="*"表示每小时，second="*"表示每秒钟。2*/a所有表示最小单位。如second=’*/2’表示最小单位为2秒，即每隔2秒。注：second=‘2’表示第二秒。3a-b所有表示[a,b]范围内包含a,b。4a-b/c所有表示[a,b]范围内每间隔c。5xthyday参数表示每个月的第x个星期y。如day=‘3rdfri’每个月的第3个星期5。6lastxday参数表示每个月最后一个星期x，如day=‘lastsun’每个月最后一个星期天。7lastday参数表示每个月的最后一天。8x,y,z所有触发任何匹配表达式;可以组合任意数量的上述表达式。如month=‘6-8,11-12’表示6，7，8，11，12月2、调度器、作业存储器、执行器详细说明2.1开始任务前，一般先对调度器进行配置。配置的内容有：1、作业存储器（jobstores）：默认的作业存储只是简单地将作业存储在内存中，但也可以存储到各种数据库中。当一个job保存到一个持久化地作业存储中时，其数据必须要被序列化（serialized），当它们被加载回来时再执行反序列化（deserialized）。非默认的作业存储不会将作业数据保存到内存中，相反，内存会作为后端存储介质在保存、加载、更新和搜索job过程中的中间人。作业存储不会在调度器（scheduler）之间共享。2、执行器（executors）：通常它们都是负责将job中指定的可调用的部分提交到线程或进程池。当job完成后，执行器会通知（notifies）调度器，由调度器随后发出（emits）一个恰当的事件（event）。3、默认添加新任务时的配置（job_defaults）。简单例子，对BackgroundScheduler进行配置：frompytzimportutcfromapscheduler.schedulers.backgroundimportBackgroundSchedulerfromapscheduler.jobstores.mongodbimportMongoDBJobStorefromapscheduler.jobstores.sqlalchemyimportSQLAlchemyJobStorefromapscheduler.executors.poolimportThreadPoolExecutor,ProcessPoolExecutor#作业存储器（jobstores）jobstores={'mongo':MongoDBJobStore(),'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}#执行器（executors）executors={'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(5)}#默认添加新任务时的配置（job_defaults）job_defaults={'coalesce':False,'max_instances':3}#在实例化时配置。#此处使用了UTC时间， 记录 混凝土 养护记录下载土方回填监理旁站记录免费下载集备记录下载集备记录下载集备记录下载 到数据库中会与中国时间相差8个小时。scheduler=BackgroundScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)#先实例化，再配置。#scheduler=BackgroundScheduler()#scheduler.configure(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)2.2作业存储器（jobstores）说明APScheduler提供了许多不同的方法来配置scheduler。你可以使用一个配置字典，或者是直接将其作为options的关键字参数。你也可以先实例化scheduler，随后再添加job和配置scheduler。有以下几种可以选择：1、MemoryJobStore：默认内存存储，没有序列化，jobs就存在内存里，重启程序，之前的job就没了。2、SQLAlchemyJobStore：所有sqlalchemy支持的数据库都可以做为backend，增删改查操作转化为对应backend的sql语句。3、MongoDBJobStore：用mongodb作任务储存。4、RedisJobStore:用redis作作任务储存。5、RethinkDBJobStore:用rethinkdb作任务储存。6、ZooKeeperJobStore：用ZooKeeper做任务储存。下面再以SQLAlchemyJobStore例子设置为存储为Mysql的：jobstores={'default':SQLAlchemyJobStore(url="mysql+pymysql://user:passwd@host/dbname?charset=utf8",table='table_name')}注：应该就是SQLAlchemy连接数据库的方式。user：用户名。passwd：密码。host：地址。dbname：数据库名。charset：编码方式。当以数据库来存储job时，如果程序终止重新启动后，默认会执行之前未执行的任务，这样会导致任务多次执行。当要避免任务重复执行，需在add_job时加入这两个参数id=‘id_name’,replace_existing=True例：scheduler.add_job(my_clock,trigger='interval',seconds=3,id='id_name',replace_existing=True)2.3执行器配置Executor在scheduler中初始化，另外也可通过scheduler的add_executor动态添加Executor。每个executor都会绑定一个alias，这个作为唯一标识绑定到Job，在实际执行时会根据Job绑定的executor找到实际的执行器对象，然后根据执行器对象执行JobExecutor的种类会根据不同的调度来选择，如果选择AsyncIO作为调度的库，那么选择AsyncIOExecutor。如果选择tornado作为调度的库，选择TornadoExecutor。如果选择启动进程作为调度，选择ThreadPoolExecutor或者ProcessPoolExecutor都可以。Executor的选择需要根据实际的scheduler来选择不同的执行器目前APScheduler支持的Executor：1、AsyncIOExecutor2、GeventExecutor3、ThreadPoolExecutor4、ProcessPoolExecutor5、TornadoExecutor6、TwistedExecutor2.4、启动和关闭调度器1、启动调度器启动调度器只需调用调度器上的start()方法。scheduler=BackgroundScheduler()…………scheduler.start()除了BlockingScheduler，非阻塞调度器都会立即返回，可以继续运行之后的代码，比如添加任务等。对于BlockingScheduler，程序则会阻塞在start()位置，所以，要运行的代码必须写在start()之前。2、关闭调度器scheduler.shutdown()默认情况下，调度器会先把正在执行的任务处理完，再关闭任务储存器和执行器。但是，如果想直接关闭，可以添加参数：scheduler.shutdown(wait=False)上述方法不管有没有任务在执行，会强制关闭调度器。3、添加工作定时任务3.1添加任务的方法有两种：1、通过调用add_job()2、通过装饰器scheduled_job()importtimefromapscheduler.schedulers.blockingimportBlockingSchedulersched=BlockingScheduler()#通过add_job()添加sched.add_job(my_job,'interval',seconds=5)#通过scheduled_job装饰器添加@sched.scheduled_job('interval',seconds=5)defmy_job():printtime.strftime('%Y-%m-%d%H:%M:%S',time.localtime(time.time()))sched.start()第一种方法是常用的；第二种方法方便，但缺点就是运行时，不能修改任务。第一种add_job()方法会返回一个apscheduler.job.Job实例，这样就可以在运行时，修改或删除任务。如果调度器还没有启动，此时添加任务，那么任务就处于一个暂存的状态。只有当调度器启动时，才会开始计算下次运行时间。还有一点要注意，如果你的执行器或任务储存器是会序列化任务的，那么这些任务就必须符合：1、回调函数必须全局可用2、回调函数参数必须也是可以被序列化的ps：内置任务储存器中，只有MemoryJobStore不会序列化任务；内置执行器中，只有ProcessPoolExecutor会序列化任务。add_job()参数说明：func：Job执行的函数trigger：apscheduler定义的触发器，用于确定Job的执行时间，根据设置的trigger规则，计算得到下次执行此job的时间，满足时将会执行。args：Job执行函数需要的位置参数kwargs：Job执行函数需要的关键字参数id：指定作业的唯一IDname：指定作业的名字misfire_grace_time：Job的延迟执行时间，例如Job的计划执行时间是21:00:00，但因服务重启或其他原因导致21:00:31才执行，如果设置此key为40,则该job会继续执行，否则将会丢弃此job。coalesce：Job是否合并执行，是一个bool值。例如scheduler停止20s后重启启动，而job的触发器设置为5s执行一次，因此此job错过了4个执行时间，如果设置为是，则会合并到一次执行，否则会逐个执行。max_instances：执行此job的最大实例数，executor执行job时，根据job的id来计算执行次数，根据设置的最大实例数来确定是否可执行。next_run_time：Job下次的执行时间，创建Job时可以指定一个时间[datetime],不指定的话则默认根据trigger获取触发时间。executor：apscheduler定义的执行器，job创建时设置执行器的名字，根据字符串你名字到scheduler获取到执行此job的执行器，执行job指定的函数。replace_existing:bool值，避免出现重复多个任务。ps:1、默认情况下，每个job同时只会有一个实例在运行。这意味着如果一个job到达计划运行时间点时，前一个job尚未完成，那么这个job最近的一次运行计划将会misfire（错过）。可以通过在添加job时指定max_instances关键字参数来设置具体job的最大实例数目，以便scheduler随后可以并发地执行它。2、有时候scheduler无法在被调度的job的计划运行时间点去执行这个job。常见的原因是这个job是在持久化的jobstore中，恰好在其打算运行的时刻scheduler被关闭或重启了。这样，这个job就被定义为misfired（错过）。scheduler稍后会检查job每个被错过的执行时间的misfire_grace_time选项（可以单独给每个job设置或者给scheduler做全局设置），以此来确定这个执行操作是否要继续被触发。这可能到导致连续多次执行。如果这个行为不符合你的实际需要，可以使用coalescing来回滚所有的被错过的执行操作为唯一的一个操作。如果对job启用了coalescing，那么即便scheduler在队列中看到这个job一个或多个执行计划，scheduler都只会触发一次。3.2删除任务如果想从调度器移除一个任务，就要从相应的任务储存器中移除它，有两种方式：1、在通过add_job()创建的任务实例上调用remove()方法。2、调用remove_job()，参数为：任务ID，任务名称。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')其它的一些任务管理方法：1、通过job实例管理。job=scheduler.add_job(myfunc,'interval',minutes=2)#添加任务job.remove()#删除任务job.pause()#暂定任务job.resume()#恢复任务job.shutdown()#关闭调度job.shutdown(wait=False)#不等待正在运行的任务2、通过调度器管理。scheduler.print_jobs()#可以快速打印格式化的任务列表，包含触发器，下次运行时间等信息。scheduler.get_jobs()#获取任务列表。scheduler.add_job(myfunc,'interval',minutes=2,id='my_job_id')#添加任务scheduler.remove_job('my_job_id')#删除任务scheduler.pause_job('my_job_id')#暂定任务scheduler.resume_job('my_job_id')#恢复任务3、修改任务通过apscheduler.job.Job.modify()或modify_job()，你可以修改任务当中除了id的任何属性。比如：job.modify(max_instances=6,name='Alternatename')如果想要重新调度任务（就是改变触发器），你能通过apscheduler.job.Job.reschedule()或reschedule_job()来实现。这些方法会重新创建触发器，并重新计算下次运行时间。比如：scheduler.reschedule_job('my_job_id',trigger='cron',minute='*/5')4、事件监听可以为scheduler绑定事件监听器（eventlisten）。Scheduler事件在某些情况下会被触发，而且它可能携带有关特定事件的细节信息。为add_listener()函数提供适当的掩码参数（maskargument）或者是将不同的常数组合到一起，可以监听特定类型的事件。可调用的listener可以通过eventobject作为参数而被调用。用add_listener(callback,mask)绑定监听事件，第一个参数是回调函数，mask是指定侦听事件类型，mask参数也可以是逻辑组合。回调函数是一个可以接收一个参数的函数。defmy_listener(event):ifevent.exception:print('Thejobcrashed:(')else:print('Thejobworked:)')print("job执行job:\ncode=>{}\njob.id=>{}\njobstore=>{}".format(event.code,event.job_id,event.jobstore))scheduler.add_listener(my_listener,EVENT_JOB_EXECUTED|EVENT_JOB_ERROR)目前APScheduler定义的EventConstant（事件类型）Description（描述）Eventclass（事件类）EVENT_SCHEDULER_STARTEDTheschedulerwasstartedSchedulerEventEVENT_SCHEDULER_SHUTDOWNTheschedulerwasshutdownSchedulerEventEVENT_SCHEDULER_PAUSEDJobprocessingintheschedulerwaspausedSchedulerEventEVENT_SCHEDULER_RESUMEDJobprocessingintheschedulerwasresumedSchedulerEventEVENT_EXECUTOR_ADDEDAnexecutorwasaddedtotheschedulerSchedulerEventEVENT_EXECUTOR_REMOVEDAnexecutorwasremovedtotheschedulerSchedulerEventEVENT_JOBSTORE_ADDEDAjobstorewasaddedtotheschedulerSchedulerEventEVENT_JOBSTORE_REMOVEDAjobstorewasremovedfromtheschedulerSchedulerEventEVENT_ALL_JOBS_REMOVEDAlljobswereremovedfromeitheralljobstoresoroneparticularjobstoreSchedulerEventEVENT_JOB_ADDEDAjobwasaddedtoajobstoreJobEventEVENT_JOB_REMOVEDAjobwasremovedfromajobstoreJobEventEVENT_JOB_MODIFIEDAjobwasmodifiedfromoutsidetheschedulerJobEventEVENT_JOB_SUBMITTEDAjobwassubmittedtoitsexecutortoberunJobSubmissionEventEVENT_JOB_MAX_INSTANCESAjobbeingsubmittedtoitsexecutorwasnotacceptedbytheexecutorbecausethejobhasalreadyreacheditsmaximumconcurrentlyexecutinginstancesJobSubmissionEventEVENT_JOB_EXECUTEDAjobwasexecutedsuccessfullyJobExecutionEventEVENT_JOB_ERRORAjobraisedanexceptionduringexecutionJobExecutionEventEVENT_JOB_MISSEDAjob’sexecutionwasmissedJobExecutionEventEVENT_ALLAcatch-allmaskthatincludeseveryeventtypeN/A -全文完-