凌的博客

您现在的位置是:首页  > 学无止境 > python

python

python 定时任务apscheduler

2019-10-06python0

1.png

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条评论