⽤Python进⾏任务程序的调度（1）
⼀直苦于python及shell脚本的串并⾏总控调度问题；虽然可以通过crontab配置进⾏⼀些不太复杂的调度，但毕竟功能有限，也显得有点low；要么就是开源调度⼯具及收费的商⽤⼯具。
⼀般在项⽬不是很庞⼤，但也有⼀点复杂性⽅⾯，在这⾥考虑⾃⼰写代码，实现程序编码的调度。
⾸先，将要调度的可并⾏程序放在相同的⽬录下；调控程序根据⽬录下的脚本⽂件和设定的并⾏参数，进⾏并⾏调度；
然后，在调度的过程中有⼀些⽇志存储，便于问题的查和发现；调度的时间设定参照crontab的格式，⼀样的⽅式配置；
再次，程序以后的优化：在图形化⽅⾯、串并⾏混合⽅⾯有很⼤的提升空间；这是⼀个初始版本，后期根据项⽬的需要进⾏相应必要的更新。
1、程序⽬录树
MacBook-Pro-Nisj:cronLake nisj$ find . -print | sed -e 's;[^/]*/;|____;g;s;____|; |;g'
.
|____scriptDir
| |____cron2
| | |____1.sh
| | |____2.py
| | |____1.py
| | |____2.sh
| |____cron1
| | |____2.py
| | |____1.py
|____threadpool.pyc
快速排序算法sql|____auxiliaryFunction.py
|____threadpool.py
|____parseCron.pyc
|____start.py
|____parseCron.py
|____cronConfig.py
|____log
| |____cronLake_run_2018-06-05.log.5
| |____cronLake_run_2018-06-05.log.2python解析json文件
| |____cronLake_run_2018-05-22.log
| |____cronLake_run_2018-06-05.log.3
| |____cronLake_run_2018-06-05.log.4
| |____cronLake_run_2018-06-05.log
| |____cronLake_run_2018-06-05.log.1
|____auxiliaryFunction.pyc
|____cronConfig.pyc
MacBook-Pro-Nisj:cronLake nisj$ pwd
/Users/nisj/PycharmProjects/BiDataProc/cronLake
MacBook-Pro-Nisj:cronLake nisj$
2、配置⽂件
/Users/nisj/PycharmProjects/BiDataProc/cronLake/cronConfig.py
# -*- coding=utf-8 -*-
cronDirMap = {
"cron1": {
"enable": "Ture",
"fileDir": "/Users/nisj/PycharmProjects/BiDataProc/cronLake/scriptDir/cron2",
"fileExt": "py",
"cron": "* * * * *",
"parallelNum": 3
},
round函数怎么用sql"cron2": {
"enable": "False",
"fileDir": "/Users/nisj/PycharmProjects/BiDataProc/cronLake/scriptDir/cron2",
"fileExt": "sh",
"cron": "* * * * *",
"parallelNum": 3
}
}
其中：⽂件以json⽅式存取，涉及程序⽬录、程序类型、并⾏个数、cron时间配置及是否⽣效5个选项；根据后期需要可以再进⾏增减。
3、解析cron格式的脚本
/Users/nisj/PycharmProjects/BiDataProc/cronLake/parseCron.py
# /usr/bin/env python
# -*- coding:utf-8 -*-
"""
1.解析 crontab 配置⽂件中的五个数间参数(分时⽇⽉周)，获取他们对应的取值范围
2.将时间戳与crontab配置中⼀⾏时间参数对⽐，判断该时间戳是否在配置设定的时间范围内
"""
import re, time, sys
def match_cont(patten, cont):
"""
正则匹配(精确符合的匹配)
Args:
patten 正则表达式
cont____ 匹配内容
Return:
True or False
"""
res = re.match(patten, cont)
if res:
return True
else:
return False
def handle_num(val, ranges=(0, 100), res=list()):
"""处理纯数字"""
val = int(val)
if val >= ranges[0] and val <= ranges[1]:
res.append(val)
return res
def handle_nlist(val, ranges=(0, 100), res=list()):
"""处理数字列表如 1,2,3,6"""
val_list = val.split(',')
for tmp_val in val_list:
tmp_val = int(tmp_val)
if tmp_val >= ranges[0] and tmp_val <= ranges[1]:
res.append(tmp_val)
res.append(tmp_val)
return res
def handle_star(val, ranges=(0, 100), res=list()):
"""处理星号"""
if val == '*':
tmp_val = ranges[0]
while tmp_val <= ranges[1]:
res.append(tmp_val)
tmp_val = tmp_val + 1
return res
def handle_starnum(val, ranges=(0, 100), res=list()):    """星号/数字组合如 */3"""
tmp = val.split('/')
val_step = int(tmp[1])
if val_step < 1:
return res
val_tmp = int(tmp[1])
while val_tmp <= ranges[1]:
res.append(val_tmp)
val_tmp = val_tmp + val_step
return res
def handle_range(val, ranges=(0, 100), res=list()):
"""处理区间如 8-20"""
tmp = val.split('-')
range1 = int(tmp[0])
range2 = int(tmp[1])
tmp_val = range1
if range1 < 0:
return res
while tmp_val <= range2 and tmp_val <= ranges[1]:        res.append(tmp_val)
tmp_val = tmp_val + 1
return res
def handle_rangedv(val, ranges=(0, 100), res=list()):    """处理区间/步长组合如 8-20/3 """
tmp = val.split('/')
range2 = tmp[0].split('-')
val_start = int(range2[0])
val_end = int(range2[1])
val_step = int(tmp[1])
if (val_step < 1) or (val_start < 0):
return res
val_tmp = val_start
while val_tmp <= val_end and val_tmp <= ranges[1]:        res.append(val_tmp)
val_tmp = val_tmp + val_step
return res
def parse_conf(conf, ranges=(0, 100), res=list()):
"""解析crontab 五个时间参数中的任意⼀个"""
# 去除空格，再拆分
conf = conf.strip(' ').strip(' ')
析构函数和构造函数的参数的区别
conf_list = conf.split(',')
other_conf = []
number_conf = []
for conf_val in conf_list:
if match_cont(PATTEN['number'], conf_val):
if match_cont(PATTEN['number'], conf_val):
# 记录拆分后的纯数字参数
number_conf.append(conf_val)
else:
# 记录拆分后纯数字以外的参数，如通配符 * , 区间 0-8, 及 0－8/3 之类
other_conf.append(conf_val)
if other_conf:
# 处理纯数字外各种参数
for conf_val in other_conf:
for key, ptn in PATTEN.items():
if match_cont(ptn, conf_val):
res = PATTEN_HANDLER[key](val=conf_val, ranges=ranges, res=res)    if number_conf:
if len(number_conf) > 1 or other_conf:
# 纯数字多于1，或纯数字与其它参数共存，则数字作为时间列表
res = handle_nlist(val=','.join(number_conf), ranges=ranges, res=res)
else:
# 只有⼀个纯数字存在，则数字为时间间隔
res = handle_num(val=number_conf[0], ranges=ranges, res=res)
return res
def parse_crontab_time(conf_string):
"""
解析crontab时间配置参数
Args:
conf_string  配置内容(共五个值：分时⽇⽉周)
取值范围分钟:0-59 ⼩时:1-23 ⽇期:1-31 ⽉份:1-12 星期:0-6(0表⽰周⽇)    Return:
crontab_range    list格式，分时⽇⽉周五个传⼊参数分别对应的取值范围
"""
time_limit = ((0, 59), (1, 23), (1, 31), (1, 12), (0, 6))
crontab_range = []
clist = []
conf_length = 5
tmp_list = conf_string.split(' ')
for val in tmp_list:
html5canva什么意思
if len(clist) == conf_length:
break
if val:
clist.append(val)
if len(clist) != conf_length:
return -1, 'config error whith [%s]' % conf_string
cindex = 0
for conf in clist:
res_conf = []
res_conf = parse_conf(conf, ranges=time_limit[cindex], res=res_conf)
if not res_conf:
return -1, 'config error whith [%s]' % conf_string
crontab_range.append(res_conf)
cindex = cindex + 1
return 0, crontab_range
# crontab时间参数各种写法的正则匹配
PATTEN = {
# 纯数字
'number': '^[0-9]+$',
# 数字列表，如 1,2,3,6
'num_list': '^[0-9]+([,][0-9]+)+$',
盐酸舍曲林吃多少片致死# 星号 *
'star': '^\*$',
# 星号/数字组合，如 */3
'star_num': '^\*\/[0-9]+$',
# 区间如 8-20
# 区间如 8-20
'range': '^[0-9]+[\-][0-9]+$',
# 区间/步长组合如 8-20/3
'range_div': '^[0-9]+[\-][0-9]+[\/][0-9]+$'
# 区间/步长列表组合，如 8-20/3,21,22,34
# 'range_div_list':'^([0-9]+[\-][0-9]+[\/][0-9]+)([,][0-9]+)+$'
}
# 各正则对应的处理⽅法
PATTEN_HANDLER = {
'number': handle_num,
'num_list': handle_nlist,
'star': handle_star,
'star_num': handle_starnum,
'range': handle_range,
'range_div': handle_rangedv
}
def getCronList(conf_string):
# 时间戳
time_stamp = int(time.time())
# 解析crontab时间配置参数分时⽇⽉周各个取值范围
res, desc = parse_crontab_time(conf_string)
if res == 0:
cron_time = desc
else:
print desc
sys, exit(-1)
# print "\nconfig:", conf_string
return {"minute": cron_time[0], "hour": cron_time[1], "day": cron_time[2], "month": cron_time[3],            "week": cron_time[4]}
# test
# if __name__ == '__main__':
#    cronDict = getCronList(conf_string='*/1 23 9-11 * 6')
#    print cronDict['week'], cronDict['month'], cronDict['day'], cronDict['hour'], cronDict['minute']
脚本从⽹上借鉴⽽来，进⾏了⼀些改动，⽅便实⽤。
4、附属功能脚本集合
/Users/nisj/PycharmProjects/BiDataProc/cronLake/auxiliaryFunction.py

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。