$ pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler


class Timing:
? ? def __init__(self, start_date, end_date, hour=None):
? ? ? ? self.start_date = start_date
? ? ? ? self.end_date = end_date
? ? ? ? self.hour = hour

? ? def cron(self, job, *value_list):
? ? ? ? """cron格式 在特定時間周期性地觸發(fā)"""
? ? ? ? # year (int 或 str) – 年，4位數(shù)字
? ? ? ? # month (int 或 str) – 月 (范圍1-12)
? ? ? ? # day (int 或 str) – 日 (范圍1-31)
? ? ? ? # week (int 或 str) – 周 (范圍1-53)
? ? ? ? # day_of_week (int 或 str) – 周內(nèi)第幾天或者星期幾 (范圍0-6 或者 mon,tue,wed,thu,fri,sat,sun)
? ? ? ? # hour (int 或 str) – 時 (范圍0-23)
? ? ? ? # minute (int 或 str) – 分 (范圍0-59)
? ? ? ? # second (int 或 str) – 秒 (范圍0-59)
? ? ? ? # start_date (datetime 或 str) – 最早開始日期(包含)
? ? ? ? # end_date (datetime 或 str) – 分 最晚結(jié)束時間(包含)
? ? ? ? # timezone (datetime.tzinfo 或str) – 指定時區(qū)
? ? ? ? scheduler = BlockingScheduler()
? ? ? ? scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,
? ? ? ? ? ? ? ? ? ? ? ? ? args=[*value_list])
? ? ? ? scheduler.start()

? ? def interval(self, job, *value_list):
? ? ? ? """interval格式 周期觸發(fā)任務(wù)"""
? ? ? ? # weeks (int) - 間隔幾周
? ? ? ? # days (int) ?- 間隔幾天
? ? ? ? # hours (int) - 間隔幾小時
? ? ? ? # minutes (int) - 間隔幾分鐘
? ? ? ? # seconds (int) - 間隔多少秒
? ? ? ? # start_date (datetime 或 str) - 開始日期
? ? ? ? # end_date (datetime 或 str) - 結(jié)束日期
? ? ? ? # timezone (datetime.tzinfo 或str) - 時區(qū)
? ? ? ? scheduler = BlockingScheduler()
? ? ? ? # 在 2019-08-29 22:15:00至2019-08-29 22:17:00期間，每隔1分30秒 運行一次 job 方法
? ? ? ? scheduler.add_job(job, 'interval', minutes=1, seconds=30, start_date=self.start_date,
? ? ? ? ? ? ? ? ? ? ? ? ? end_date=self.end_date, args=[*value_list])
? ? ? ? scheduler.start()

? ? @staticmethod
? ? def date(job, *value_list):
? ? ? ? """date格式 特定時間點觸發(fā)"""
? ? ? ? # run_date (datetime 或 str) - 作業(yè)的運行日期或時間
? ? ? ? # timezone (datetime.tzinfo 或 str) ?- 指定時區(qū)
? ? ? ? scheduler = BlockingScheduler()
? ? ? ? # 在 2019-8-30 01:00:01 運行一次 job 方法
? ? ? ? scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=[*value_list])
? ? ? ? scheduler.start()

if __name__ == '__main__':
? ? file_list = ["test_shiyan.py", "MeetSpringFestival.py"]
? ? # run_py(file_list)
? ? case_list = ["test_case_01", "test_case_02"]
? ? # run_case(test_sample, case_list)
? ? dingDing_list = [2, case_list, test_sample]
? ? # run_dingDing(*dingDing_list)
? ? Timing('2022-02-15 00:00:00', '2022-02-16 00:00:00', '0-23').cron(run_dingDing, *dingDing_list)

def cron(self, job, *value_list):
? ? ? ? """cron格式 在特定時間周期性地觸發(fā)"""
?? ??? ?scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,
?? ??? ? ? ? ? ? ? ? ? ? ? ? ? ? ?args=[*value_list])

 class Config(object):
 
    SCHEDULER_JOBSTORES = {
        'default': RedisJobStore(db=3,host='0.0.0.0', port=6378,password='******'),
    }
 
    SCHEDULER_EXECUTORS = {
        'default': {'type': 'processpool', 'max_workers': 50}  #用進(jìn)程池提升任務(wù)處理效率
    }
 
    SCHEDULER_JOB_DEFAULTS = {
        'coalesce': True,   #積攢的任務(wù)只跑一次
        'max_instances': 1000, #支持1000個實例并發(fā)
       'misfire_grace_time':600 #600秒的任務(wù)超時容錯
    }
 
    SCHEDULER_API_ENABLED = True 

 
{
"id": "9586_site_status",
"name": "9586_site_status",
"func": "monitor_scheduler:monitor_site_status",
"args": [
9586,
"http://sl.jxcn.cn/",
1000,
100,
200,
"",
0,
2
],
"kwargs": {},
"trigger": "interval",
"start_date": "2018-09-14T00:00:00+08:00",
"end_date": "2018-12-31T00:00:00+08:00",
"minutes": 15,
"misfire_grace_time": 10,
"max_instances": 3000,
"next_run_time": "2018-10-24T18:00:00+08:00"
} 

 scheduler.modify_job(func=func, id=id, args=args, trigger=trigger, minutes=minutes,start_date=start_date,end_date=end_date,misfire_grace_time=600)