Python实现定时任务利器之apscheduler使用详解_python_

#encoding:utf-8 from apscheduler.schedulers.blocking import BlockingScheduler import datetime def sch_test(): now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print('时间：{}, 测试apscheduler'.format(now)) task = BlockingScheduler() task.add_job(func=sch_test, trigger='cron', second='*/10') task.start() 

# 使用run_date指定运行时间 task.add_job(func='sch_test', trigger='date', run_date=datetime.datetime(2022 ,10 , 8, 16, 1, 30)) # 或者用next_run_time task.add_job(func=sch_test,trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3)) 

# encoding:utf-8 from apscheduler.schedulers.blocking import BlockingScheduler import datetime def sch_test(job_type): now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print('时间：{}, {}测试apscheduler'.format(now, job_type)) task = BlockingScheduler() task.add_job(func=sch_test, args=('一次性任务',),trigger='date', next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=3)) task.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5') task.add_job(func=sch_test, args=('循环任务',), trigger='interval', seconds=3) task.start() 

from apscheduler.schedulers.blocking import BlockingScheduler def sch_test(job_type): now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print('时间：{}, {}测试apscheduler'.format(now, job_type)) sched = BlockingScheduler() # 使用mysql存储任务 sql_url = 'mysql+pymysql://root:root@localhost:3306/db_name?charset=utf8' sched.add_jobstore('sqlalchemy',url=sql_url) # 添加任务 sched.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5') sched.start() 

from apscheduler.schedulers.blocking import BlockingScheduler import datetime from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR import logging # logging日志配置打印格式及保存位置 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='sche.log', filemode='a') def log_listen(event): if event.exception : print ( '任务出错，报错信息：{}'.format(event.exception)) else: print ( '任务正常运行...' ) def sch_test(job_type): now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print('时间：{}, {}测试apscheduler'.format(now, job_type)) print(1/0) sched = BlockingScheduler() # 使用mysql存储任务 sql_url = 'mysql+pymysql://root:root@localhost:3306/db?charset=utf8' sched.add_jobstore('sqlalchemy',url=sql_url) # 添加任务 sched.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5') # 配置任务执行完成及错误时的监听 sched.add_listener(log_listen, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) # 配置日志监听 sched._logger = logging sched.start() 

from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR import logging import logging.handlers import os import datetime class LoggerUtils(): def init_logger(self, logger_name): # 日志格式 formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') log_obj = logging.getLogger(logger_name) log_obj.setLevel(logging.INFO) # 设置log存储位置 path = '/data/logs/' filename = '{}{}.log'.format(path, logger_name) if not os.path.exists(path): os.makedirs(path) # 设置日志按照时间分割 timeHandler = logging.handlers.TimedRotatingFileHandler( filename, when='D', # 按照什么维度切割， S:秒，M：分，H:小时，D:天，W:周 interval=1, # 多少天切割一次 backupCount=10 # 保留几天 ) timeHandler.setLevel(logging.INFO) timeHandler.setFormatter(formatter) log_obj.addHandler(timeHandler) return log_obj class Scheduler(LoggerUtils): def __init__(self): # 执行器设置 executors = { 'default': ThreadPoolExecutor(10), # 设置一个名为“default”的ThreadPoolExecutor，其worker值为10 'processpool': ProcessPoolExecutor(5) # 设置一个名为“processpool”的ProcessPoolExecutor，其work
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：如何将bag文件批量转成mp4_python_
                                                                
下一篇：Python os.listdir与os.walk实现获取路径详解_python_