python 定时任务apscheduler
2019-10-06 python 2035
import os import apscheduler from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger import logging import time class Shelduler: task_index = 0 wait_index = 0 def __init__(self): self.scheduler = BackgroundScheduler() logging.basicConfig(level=logging.ERROR, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S', filename='log1.txt', filemode='a') def run(self): print("run init") scheduler = self.scheduler mytime = "00:00:00" times = mytime.split(":") step = 5 # 间隔5秒钟执行一次 trigger = IntervalTrigger(seconds=step) scheduler.add_job(self.mytask, trigger) scheduler.add_listener(self.myListener, apscheduler.events.EVENT_JOB_EXECUTED | apscheduler.events.EVENT_JOB_ERROR) trigger = CronTrigger(day_of_week="0-6", hour=int(times[0]), minute=int(times[1]), second=int(times[2])) scheduler.add_job(self.mytask2, trigger) scheduler.start() print('按下 Ctrl+{0} 退出'.format("Break" if os.name !='nt' else 'C')) try: # 其他任务 独立的线程执行 while True: time.sleep(2) self.wait_index += 1 print("\r" + "waitting" + "..." * self.wait_index, end="OK") self.task_index += 1 if self.task_index % 10 == 0: # 执行10次 清空屏幕 os.system("cls") self.task_index = 0 self.wait_index = 0 except (KeyboardInterrupt, SystemExit): scheduler.shutdown() print('\n退出任务!') def myListener(self, ev): if ev.exception: logging.info('\n%s error.', str(ev.exception)) else: print("执行成功") def mytask(self): print("\nmy task1") self.wait_index = 0 def mytask2(self): print("\nmy task2") if __name__ == "__main__": shed = Shelduler() shed.run()
很赞哦! (0)
相关文章
文章评论
-
-
-
0条评论