python 实现后台cron_tab
- 2020 年 1 月 10 日
- 筆記
后台cron_table管理 PHP安装扩展也可以真正的多线程,fork新的进程,php ,python 2种方式性能上差别多少不知道. 基于python 强大,简洁选用python 做核心功能. 使用mysql表存储cron任务,python守护进程每分钟读取MYSQL, 多线程分发fork子进程方式执行PHP代码. 具体流程: 1.cron 每分钟执行cron_table.py . 2.cron_table.py 读取cron 任务表,把符合当前时间执行的cron 记录状态更新为runing 3.cron_table_log 插入一条记录,cron_id,start_time 4.采用多线程方式fork守护子进程,等待子进程执行完, 5.子进程执行完子线程修改cron 任务的状态为no runing,更新log记录表的完成时间和子进程的输出信息. cron_table功能: 1.web后台能添加修改cron 任务,可视化的操作.脱离需要操作服务器才能控制crontab 2.及时的记录每一条cron当前的执行状态,时间,以及历史执行状态记录..对cron任务 何时执行,执行时长,执行返回信息,等全面信息监控.对CRON 脚本优化,排查异常CRON等有直接的帮助. 快速找到死循环CRON,性能低下的CRON程序. 3.允许CRON 任务独占,许多任务是不能同时跑多次,做的独占机制能有效的避免上个周期跑的CRON 还没结束, 这个周期到了,继续跑.. 4.可以设置cron最大运行时间 5.完全脱离cron管理和cron任务业务代码的联系. 有需要还允许后台控制kill cron 任务,可以不用shell 命令终止,发邮件,短信通知 相对比现在项目里的cron记录方式,在构造函数,析构函数做file_put_contents低效操作. 这种方式重大缺陷:CRON死循环不做析构函数写日志,CRON业务代码出错,整个CRON 不执行无法记录等..
#!/usr/bin/python #coding=utf-8 #引入 MySQL 模組 import MySQLdb, re, time, sys, subprocess, threading def get_struct_time(time_stamp_int): """ 按整型时间戳获取格式化时间 分 时 日 月 周 Args: time_stamp_int 为传入的值为时间戳(×××)，如：1332888820 经过localtime转换后变成 time.struct_time(tm_year=2012, tm_mon=3, tm_mday=28, tm_hour=6, tm_min=53, tm_sec=40, tm_wday=2, tm_yday=88, tm_isdst=0) Return: list____返回 分 时 日 月 周 """ st_time = time.localtime(time_stamp_int) return [st_time.tm_min, st_time.tm_hour, st_time.tm_mday, st_time.tm_mon, st_time.tm_wday] def get_strptime(time_str, str_format): """从字符串获取 整型时间戳 Args: time_str 字符串类型的时间戳 如 '31/Jul/2013:17:46:01' str_format 指定 time_str 的格式 如 '%d/%b/%Y:%H:%M:%S' Return: 返回10位整型(int)时间戳，如 1375146861 """ return int(time.mktime(time.strptime(time_str, str_format))) def get_str_time(time_stamp, str_format='%Y%m%d%H%M'): """ 获取时间戳, Args: time_stamp 10位整型(int)时间戳，如 1375146861 str_format 指定返回格式，值类型为 字符串 str Rturn: 返回格式 默认为 年月日时分，如2013年7月9日1时3分 :201207090103 """ return time.strftime("%s" % str_format, time.localtime(time_stamp)) 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) 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): #记录拆分后的纯数字参数 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: 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 def time_match_crontab(crontab_time, time_struct): """ 将时间戳与crontab配置中一行时间参数对比，判断该时间戳是否在配置设定的时间范围内 Args: crontab_time____crontab配置中的五个时间（分 时 日 月 周)参数对应时间取值范围 time_struct____ 某个整型时间戳，如：1375027200 对应的 分 时 日 月 周 Return: tuple 状态码, 状态描述 """ cindex = 0 for val in time_struct: if val not in crontab_time[cindex]: return 0, False cindex = cindex + 1 return 0, True #crontab时间参数各种写法 的 正则匹配 PATTEN = { #纯数字 'number':'^[0-9]+$', #数字列表,如 1,2,3,6 'num_list':'^[0-9]+([,][0-9]+)+$', #星号 * 'star':'^*$', #星号/数字 组合，如 */3 'star_num':'^*/[0-9]+$', #区间 如 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 mysqlConnect(): return MySQLdb.connect(host="192.168.51.129", user="root", passwd="123456", db="model") def runCronTable(ns,control,method,id,logid): child = subprocess.Popen('/usr/local/php-5.4.20/bin/php -c /usr/local/php-5.4.20/php.ini index.php --ns=%s --control=%s --method=%s --id=%s' % (ns,control,method,id),shell=True,stdout = subprocess.PIPE) info,returnCode = child.communicate() db = mysqlConnect() cursor = db.cursor(MySQLdb.cursors.DictCursor) cursor.execute('update cron_table set done_time=%s,run_status=0 where id=%s' , (time.time(),id)) cursor.execute('update cron_table_log set done_time=%s,info=%s where id=%s' , (time.time(),info,logid)) cursor.close() db.close() return True if __name__ == '__main__': #連接到 MySQL db = mysqlConnect() cursor = db.cursor(MySQLdb.cursors.DictCursor) #執行 SQL 語句 cursor.execute("SELECT * FROM cron_table where status=0 and run_status=0") result = cursor.fetchall() threads = [] now = time.strftime('%M %H %d %m %w',time.localtime(time.time())) for r in result: res, desc = parse_crontab_time(r["cycle"]) if res == 0: cron_time = desc else: cron_time = 0 time_stamp = int(time.time()) #解析 时间戳对应的 分 时 日 月 周 time_struct = get_struct_time(time_stamp) match_res,isRun = time_match_crontab(cron_time, time_struct) if isRun: now = time.time() print "%s.." % r["id"] # 锁住cron 任务 cursor.execute("update cron_table set run_status=%s where id=%s" , (now,r["id"])) # 添加日志 cursor.execute("insert cron_table_log (cron_id,start_time,done_time,info) VALUES (%s,%s,0,'')" , (r["id"],now)) thd = threading.Thread(target=runCronTable,args=(r["ns"],r["control"],r["method"],r["id"],int(db.insert_id()))) thd.setDaemon(True) thd.start() threads.append(thd) db.close() cursor.close() for t in threads: t.join() print "nExiting Main Threadn"
MYSQL 表结构
CREATE TABLE `cron_table` ( `id` int(10) unsigned NOT NULL auto_increment, `ns` varchar(100) NOT NULL default '' COMMENT '命名空间', `control` varchar(50) NOT NULL default '' COMMENT '控制器', `method` varchar(50) NOT NULL default '' COMMENT '方法', `cycle` varchar(50) NOT NULL COMMENT '运行周期', `run_status` int(10) unsigned NOT NULL default '0' COMMENT '运行状态', `done_time` int(10) unsigned NOT NULL default '0' COMMENT '完成时间', `status` tinyint(4) unsigned NOT NULL default '1' COMMENT '开启关闭', `desc` varchar(500) NOT NULL default '' COMMENT 'CRON描述', PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `cron_table_log` ( `id` int(10) unsigned NOT NULL auto_increment, `cron_id` int(10) unsigned NOT NULL default '0' COMMENT 'cron_id', `start_time` int(10) unsigned NOT NULL default '0' COMMENT '开始执行时间', `done_time` int(10) unsigned NOT NULL default '0' COMMENT '脚本执行完成时间', `info` varchar(3000) NOT NULL default '' COMMENT '脚本输出', PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
