没有使用apscheduler。自己实现了一个定时任务，目前可满足自己项目的需要。
采用跟linux类似的定时任务命令，不同的是这里不支持星期，增加了秒定时任务。
以下是源码。
class Cron:
def __init__(self):
self.log = logClass()
self.now_tm = int(time.time())
self.text_month = 'month'
self.text_day = 'day'
self.text_hour = 'hour'
self.text_min = 'min'
self.text_sec = 'sec'
def getNowDetailTime(self):
date = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# self.log.debug('now_date:{}'.format(int(date)))
return date
def getYear(self):
date = datetime.datetime.now().strftime("%Y")
# self.log.debug('now_year:{}'.format(int(date)))
return int(date)
def getMonth(self):
date = datetime.datetime.now().strftime("%m")
# self.log.debug('now_month:{}'.format(int(date)))
return int(date)
def getDay(self):
date = datetime.datetime.now().strftime("%d")
# self.log.debug('now_day:{}'.format(int(date)))
return int(date)
def getHour(self):
date = datetime.datetime.now().strftime("%H")
# self.log.debug('now_hour:{}'.format(int(date)))
return int(date)
def getMinute(self):
date = datetime.datetime.now().strftime("%M")
# self.log.debug('now_minute:{}'.format(int(date)))
return int(date)
def getSecond(self):
date = datetime.datetime.now().strftime("%S")
# self.log.debug('now_second:{}'.format(int(date)))
return int(date)
def getIntervalList(self,It,n,start):
mlist = []
if It==self.text_month or It==self.text_hour:
for x in range(start, 13):
if x % n == 0:
mlist.append(x)
elif It==self.text_day:
for x in range(start, GetCurrentMonthDay()+1):
if x % n == 0:
mlist.append(x)
elif It==self.text_min or It==self.text_sec:
for x in range(start, 61):
if x % n == 0:
mlist.append(x)
# self.log.debug('type:[{}],n:[{}],mlist:{}'.format(It,n,mlist))
return mlist
def cron_controller(self,cron_str):
"""
测试用例：
*,*,*,*,*
*,*,*,45,*
*,*,*,45,0
*,*,*,45,*/3
*,*,*,*/1,*
*,*,*,*/2,*
*,*,*,*/4,*
*,*,*/3,*/4,*
*,*,*/3,*/4,0
共计三种类型，有斜杠、全是* ，纯数字
:param cron_str:
:return:
"""
r = {}
rdict = {
self.text_month:0,
self.text_day:0,
self.text_hour:0,
self.text_min:0,
self.text_sec:0,
'is_pass':0,
'cron_text':''
}
cron = str(cron_str).split('|') if '|' in str(cron_str) else 0
if isinstance(cron,list):
rdict['cron_text']=cron[1]
cron = cron[0]
else:
cron = cron_str
d = str(cron).split(',') if ',' in str(cron) else 0
cron_dict = {}
debug = {}
if d != 0 and len(d) == 5:
cron_dict['month']=d[0]
cron_dict['day']=d[1]
cron_dict['hour']=d[2]
cron_dict['min']=d[3]
cron_dict['sec']=d[4]
r['cron']=cron_dict
for stype,son in cron_dict.items():
res = 0
sb = 0
sa = 0
if '*' == son:
res=1
elif '/' in son:
# 针对*/2 的情况
s = str(son).split('/')
if len(s)==2:
sb = int(s[0]) if '*' not in s[0] else 1
sa = int(s[1]) if '*' not in s[1] else 1
# self.log.debug('sb:{},sa:{}'.format(sb,sa))
elif son.isdigit():
# 针对 指定时间执行的情况 *，*，*，45，*
res=2
sa=int(son)
if res==0 and sb and sa: # sa=0的情况要单独处理,这个节点不允许进入执行
if stype==self.text_month:
self.now_month = self.getMonth()
if self.now_month%sb==0 and self.now_month %sa==0:
rdict[self.text_month]=1
if stype==self.text_day:
self.now_day = self.getDay()
if self.now_day % sb == 0 and self.now_day %sa==0:
rdict[self.text_day] = 1
if stype==self.text_hour:
self.now_hour = self.getHour()
if self.now_hour !=0:
if self.now_hour % sb == 0 and self.now_hour %sa==0:
rdict[self.text_hour] = 1
else:
# 当小时等于=0 时，也就是24点，此时只要能被24 整除的定时任务配置都可以执行，故设置为通过，以此类推 时分秒 都有这种情况
if 24 %sa ==0:
rdict[self.text_hour] = 1
if stype==self.text_min:
self.now_min = self.getMinute()
if self.now_min==0:
# 当小时等于=0 时，*/1 */2 两种情况都有可能会在0点执行，故设置为通过，以此类推 时分秒 都有这种情况
if 60%sa==0:
rdict[self.text_min] = 1
else:
if self.now_min % sb == 0 and self.now_min %sa==0:
rdict[self.text_min] = 1
if stype==self.text_sec:
self.now_sec = self.getSecond()
if self.now_sec==0:
# 当小时等于=0 时，*/1 */2 两种情况都有可能会在0点执行，故设置为通过，以此类推 时分秒 都有这种情况
self.log.debug('sec:{},sa:{},res:{}'.format(self.now_sec,sa,60%sa))
if 60%sa==0:
rdict[self.text_sec] = 1
else:
if self.now_sec % sb == 0 and self.now_sec %sa == 0:
rdict[self.text_sec] = 1
elif res==2: # 此时定时任务为整数，sa=0的情况针对时分秒要单独处理
if stype==self.text_month:
self.now_month = self.getMonth()
if sa !=0:
if self.now_month == sa:
rdict[self.text_month]=1
if stype==self.text_day:
if sa !=0:
self.now_day = self.getDay()
if self.now_day == sa:
rdict[self.text_day] = 1
if stype==self.text_hour:
self.now_hour = self.getHour()
if sa != 0:
if self.now_hour == sa:
rdict[self.text_hour] = 1
else:
if self.now_hour==0:
rdict[self.text_hour] = 1
if stype==self.text_min:
self.now_min = self.getMinute()
if sa != 0:
if self.now_min==sa:
rdict[self.text_min] = 1
else:
if self.now_min==0:
rdict[self.text_min] = 1
if stype==self.text_sec:
self.now_sec = self.getSecond()
if sa != 0:
if self.now_sec==sa:
rdict[self.text_sec] = 1
else:
if self.now_sec==0:
rdict[self.text_sec] = 1
else:
debug['res_0']=1
if stype==self.text_month:
rdict[self.text_month]=1
if stype==self.text_day:
rdict[self.text_day]=1
if stype==self.text_hour:
rdict[self.text_hour]=1
if stype==self.text_min:
rdict[self.text_min]=1
if stype==self.text_sec:
rdict[self.text_sec]=1
# self.log.debug('type:{},son:{},res:{},res_dict:{}'.format(stype, son, res,rdict))
# self.log.debug('***** ok, rdict:{}'.format(rdict))
if rdict[self.text_month] and rdict[self.text_day] and rdict[self.text_hour] and rdict[self.text_min] and rdict[self.text_sec]:
rdict['is_pass'] = 1
else:
self.log.error('定时任务参数错误：{}'.format(cron))
r['result']=rdict
if rdict['is_pass']==1:
self.log.debug(r)
return r
def debug(self):
self.getNowDetailTime()
self.getYear()
self.getMonth()
self.getDay()
self.getHour()
self.getMinute()
self.getSecond()
cr = Cron()
# 将参数传递给装饰器，符合条件将后调用函数
def cron(cron_p):
def func_wrapper(func):
@wraps(func)
def return_wrapper(*args, **wkargs):
# 函数通过装饰起参数给装饰器传送参数
res = cr.cron_controller(cron_p)
if res['result']['is_pass']:
func(*args, **wkargs)
return return_wrapper
return func_wrapper
使用方法：
import multiprocessing
@cron('*,*,*,*/5,0|auditUploadMonitor') # 竖线前面是定时任务，后面是说明
def auditUploadMonitor():
aum= AuditUploadMonitor()
aum.audit_overview()
def main():
fun_list = [oneapstatistical, a,b,c,d]
# fun_list=[authTotal,auditUploadMonitor]
for i in fun_list:
pool = multiprocessing.Process(target=i)
pool.start()
while True:
time.sleep(1)
main()
原理：
启动常驻主进程，然后每一秒轮训定时任务，如歌符合要求，则执行方法，否则不执行。
网友评论