定時任務觸發方式有幾種類型，日常的工作中，研發同學運用比較多的就是cron方式
查了一下APScheduler框架內支援多種定時任務方式
先安裝apscheduler模組
$ 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格式 在特定时间周期性地触发"""
# year (int 或 str) – 年，4位数字
# month (int 或 str) – 月 (范围1-12)
# day (int 或 str) – 日 (范围1-31)
# week (int 或 str) – 周 (范围1-53)
# day_of_week (int 或 str) – 周内第几天或者星期几 (范围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) – 分 最晚结束时间(包含)
# timezone (datetime.tzinfo 或str) – 指定时区
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格式 周期触发任务"""
# weeks (int) - 间隔几周
# days (int) - 间隔几天
# hours (int) - 间隔几小时
# minutes (int) - 间隔几分钟
# seconds (int) - 间隔多少秒
# start_date (datetime 或 str) - 开始日期
# end_date (datetime 或 str) - 结束日期
# timezone (datetime.tzinfo 或str) - 时区
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格式 特定时间点触发"""
# run_date (datetime 或 str) - 作业的运行日期或时间
# timezone (datetime.tzinfo 或 str) - 指定时区
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)把run_dingDing()的函數我們放在已經封裝好的Timing().cron(run_dingDing，*dingDing_list)內，那麼run_dingDing()內的參數我們透過元組的方式傳入
就是我們上面寫的這裡能看到
def cron(self, job, *value_list):
"""cron格式 在特定时间周期性地触发"""
scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,
args=[*value_list])時間範圍的填寫我放在了Timing()初始化內，看著舒服一點
在運行Timing().cron()後就可以觸發定時了，但是必須要開著電腦才可以，等後面開始研究平台，存放在伺服器內就美吱吱了～
apscheduler 運作過程中出現類似如下報錯：
##針對該問題百度基本上是指不上了，google到了關鍵配置，但仍然出現該報錯，於是繼續找資料，刨根問底這到是什麼鬼問題導致的。Run time of job "9668_hack (trigger: interval[1:00:00], next run at: 2018- 10-29 22:00:00 CST)" was missed by 0:01:47.387821Run time of job "9668_index (trigger: interval[0:30:00], next run at: 2018-10-29 21:309 21:21: 00 CST)" was missed by 0:01:47.392574Run time of job "9669_deep (trigger: interval[1:00:00], next run at: 2018-10-29 22:00:00 CST)" 0:01:47.397622Run time of job "9669_hack (trigger: interval[1:00:00], next run at: 2018-10-29 22:00:00 CST)" was missed by 0:01:47. of job "9669_index (trigger: interval[0:30:00], next run at: 2018-10-29 21:30:00 CST)" was missed by 0:01:47.407996 )" was missed by 0:01:47.407996 )" was missed by 0:01:47.407996
#coalesce：當由於某種原因導致某個job積累了好幾次沒有實際運行（比如說系統掛了5分鐘後恢復，有一個任務是每分鐘跑一次的，按道理說這5分鐘內本來是「計劃」運行5次的，但實際沒有執行），如果coalesce為True，下次這個job被submit給executor時，只會執行1次，也就是最後這次，如果為False，那麼會執行5次（不一定，因為還有其他條件，看後面misfire_grace_time的解釋）
max_instance：就是說同一個job同一時間最多有幾個實例再跑，例如一個耗時10分鐘的job，被指定每分鐘運行1次，如果我們max_instance值為5，那麼在第6~10分鐘上，新的運行實例不會被執行，因為已經有5個實例在跑了
：設想和上述coalesce類似的場景，如果一個job本來14:00有一次執行，但是由於某種原因沒有被調度上，現在14:01了，這個14:00的運行實例被提交時，會檢查它預訂運行的時間和當下時間的差值（這裡是1分鐘），大於我們設定的30秒限制，那麼這個運行實例不會被執行。
15分鐘一次的任務，misfire_grace_time 設定100秒，在0:06分的時候提示：
Run time of job "9392_index (trigger: interval[0:15:00], next run at: 2018-10-27 00:15:00 CST)" was missed by 0:06:03.931026##解釋：
本來應該在0：00執行的任務，某種原因沒有被調度，提示下次執行（ 0：15）與目前差了6分鐘（閾值100秒），所以0：15的時候將不會運行
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} #用进程池提升任务处理效率
}
SCHEDULER_JOB_DEFAULTS = {
'coalesce': True, #积攒的任务只跑一次
'max_instances': 1000, #支持1000个实例并发
'misfire_grace_time':600 #600秒的任务超时容错
}
SCHEDULER_API_ENABLED = True我本以为这样应该就没什么问题了，配置看似完美，但是现实是残忍的，盯着apscheduler日志看了一会，熟悉的“was missed by”又出现了，这时候就需要怀疑这个配置到底有没有生效了，然后发现果然没有生效，从/scheduler/jobs中可以看到任务：
{
"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"
}可以看到任务中默认就有misfire_grace_time配置，没有改为600，折腾一会发现修改配置，重启与修改任务都不会生效，只能修改配置后删除任务重新添加（才能把这个默认配置用上），或者修改任务的时候把这个值改掉
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)
然后就可以了？图样图森破，missed 依然存在。
其实从后来的报错可以发现这个容错时间是用上的，因为从执行时间加上600秒后才出现的报错。
那么还是回到这个超时根本问题上，即使容错时间足够长，没有这个报错了，但是一个任务执行时间过长仍然是个根本问题，所以终极思路还在于如何优化executor的执行时间上。
当然这里根据不同的任务处理方式是不一样的，在于各自的代码了，比如更改链接方式、代码是否有冗余请求，是否可以改为异步执行，等等。
而我自己的任务解决方式为：由接口请求改为python模块直接传参，redis链接改为内网，极大提升执行效率，所以也就控制了执行超时问题。
以上是python apscheduler cron定時任務觸發介面自動化巡檢怎麼實現的詳細內容。更多資訊請關注PHP中文網其他相關文章！
