jenkins定时任务Cron表达式简介
jenkins是目前市场上最主流最好用的持续集成平台,几乎每个科技公司都在使用:
jenkins定时任务一直以来非常受欢迎,通过简单的命令,既可以实现几乎无所不能的定时需求。
命令举例如下:
背景
笔者之前所在的企业里,经常会有很多内部自动化平台或者自动跑批等平台开发出来,之前这些平台的定时任务都是交给公司内唯一的一套jenkins来执行触发。
但可能是因为要支持的任务太多,jnekins表现越来越卡了,甚至一度宕机。
那天,运维同事终于忍不了了,叫我们自己去搭建jenkins,公司内的唯一一套jenkins不给我们随便玩了...
于是,各组纷纷搭建自己的jenkins服务,但因为对运维方面的知识并不专业,加上也没有专用的好服务器来运行jenksin,导致服务异常不稳。后来我们总结了原因,领导说,一个平台所需要的支撑服务越多,链路越长,就越不稳定。我们之前成功跨越了组件化阶段,实现了业务/技术中台,虽然省心。但现在来看,反而可能会牵一发动全身,非常被动也不灵活。
于是他们决定要把部分不稳定的支持服务回退到组件化阶段!这其中,就包括了jenkins服务。
用领导的意思说就是,我们很多内部自动化平台,其实仅仅就是用一下jenkins的定时任务罢了,没必要单独部署一整套jenkins还要占个服务器。既然公司共用的jenkins不给我们玩了,那我们自己做一套jenkins的定时任务工具就好,然后最好做成可pip install 直接下载的第三方库,以后用的时候直接导入就用,轻便快捷,还稳定。
乍一听,不就是一个时间定时函数么?能有啥难得,于是大家纷纷表示可以搞。但下一个瞬间,大家突然发现好像不对....
jenkins的定时表达式,读起来非常简单,但如果解析并实现这个底层貌似有些难搞呀?
于是在领导问谁来负责的时候,不出意外的,大家朝着我伸出了他们那罪恶的食指....
我的思考
要解析这个表达式,并成功按照指定时间触发。我觉得最好的办法,就是根据当前时间,计算出下一次要触发的时间,然后一个轮询等到时间就执行就好。
结果里面居然引发出来好多难点。比如如何计算出下一次触发的时间,如何要节省的去轮询?多少秒一次?允许误差是几秒?是否需要做个高幂等性锁来防止触发两次?等等问题,让我感到压力。
不过好在,把难题拆分后,还是可以坚持搞下去的。否则我真的当场GG了。
正文
为了贯彻我们领导的尽量精简链路思想,所以我选择用python,并且只用一个函数来实现。函数接收jenkins的定时任务表达式字符串为唯一入参。返回为距离现在最近一次要触发的时间戳。因为时间戳在我们具体开发中才是易于使用的而非给人看的日期格式。
简单易用!
def get_next_time(s='* * * * *'): if s == '* * * * *': return time.time() + 60 s = s.split(' ') for i in range(len(s)): fg = '' if '/' in s[i]: fg = int(s[i].split('/')[1]) s[i] = s[i].split('/')[0] if '-' in s[i]: s[i] = list(range(int(s[i].split('-')[0]),int(s[i].split('-')[1])+1)) elif ',' in s[i]: s[i] = [int(j) for j in s[i].split(',')] elif '*' in s[i]: s[i] = [list(range(0,60)),list(range(0,24)),list(range(1,32)),list(range(1,13)),list(range(1,8))][i] else: s[i] = [int(s[i])] if fg: s[i] = list(range(s[i][0],s[i][-1]+1,fg)) tmp = list(itertools.product(s[0],s[1],s[2],s[3]) ) all = [] for i in tmp: t = '%s-%s-%s %s:%s:00'%(time.localtime(time.time())[0],i[3],i[2],i[1],i[0]) try: s_t = time.strptime(t, "%Y-%m-%d %H:%M:%S") except: continue mkt = int(time.mktime(s_t)) if mkt > time.time(): rq = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(mkt)) dw = datetime.strptime(rq.split(' ')[0],'%Y-%m-%d').isoweekday() if dw in s[-1]: all.append(mkt) all.sort() if all: return all[0] else: return None
简单来讲解一下(虽然这是第一版,并不完美,也没有完全解析的了jenkins表达式。但是已经足够我们组内使用了!)
思路:
先把表达式拆分成每一段:分、时、日、月、星期
然后把每一段符合的时间数字,做成一个列表。最终整个表达式字符串就变成了一个二维大列表。
然后这里每个列表的不同组合,就是所有符合的时间节点了。
我们再从者所有时间节点中,找到距离现在最近的下一个执行时间节点即可。
但是要进行时间排序的话,干脆就全变成时间戳,然后通过sort方法自动排序快。然后截取现在时间戳之后的片段,再拿到最小的那个,就是最终结果了!就是下一个要执行触发的时间节点。
首先,就是把这个表达式按照空格进行了拆解,每一部分再分析其所属类别:/ - , * 其他等等。
针对每个部分,都安排了算法进行解析。(注意,尽量避免出现60分或24时,可用0代表,也可以自行修改设计。我们组有自己业务上的考虑才故意这样写...)
对每个部分的可能值都放进了一个列表。
所以中间就变成了一个二维列表。
举例:希望每天的6:30 启动触发任务
经过前半部分的分段解析后,二维列表如下:
这就是当前,可以触发的所有时间日期节点。
然后我们再在里面找出距离现在时间最近的一个未来时间戳。
这个未来时间都是放在all变量中,all变量经过.sort()后会自动排序,从小到大。所以用时间戳而不是具体日期格式来计算出最近节点是可靠并简单的。
所以我们返回这个all[0]为最终输出。
结果如下:
那么这个时间到底对不对呢?
看一下我限制写这篇文章的时间:是6.5号深夜了
然后我们把这个时间戳翻译成日期:(见证奇迹的时刻来了!!!)
计算出的最近触发时间节点是:6.6号的早上6:30
完全正确!非常nice!
好,到此算是初步完成了领导的要求了。
虽然领导表示,你这个算法貌似不够完善,也不够健壮,更不符合pep8风格标准,甚至可能还存在着边界值bug。
但是整个的这个思路设计,非常好啊!非常值得组内推广!
于是,上传成了pypi的一个工具:python_jenkins_monitor
以后组员下载方便了,直接命令行:pip3 install python_jenkins_monitor 即可。
虽然后续该代码被领导给优化过了(加了好多标点符号和注释既视感),但是狗领导不给更新到pypi了。
大家如果想用的,自行优化下,或者留言讨论下吧~