做一个可通过jenkins定时任务Cron表达式设置的python定时函数

简介: 用python代码,来解析jenkins定时任务表达式,并获取最近的执行任务时间戳

image.png

jenkins定时任务Cron表达式简介

jenkins是目前市场上最主流最好用的持续集成平台,几乎每个科技公司都在使用:

image.png

jenkins定时任务一直以来非常受欢迎,通过简单的命令,既可以实现几乎无所不能的定时需求。

命令举例如下:

image.png

背景

笔者之前所在的企业里,经常会有很多内部自动化平台或者自动跑批等平台开发出来,之前这些平台的定时任务都是交给公司内唯一的一套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 启动触发任务

image.png


经过前半部分的分段解析后,二维列表如下:

image.png

这就是当前,可以触发的所有时间日期节点。

然后我们再在里面找出距离现在时间最近的一个未来时间戳。

这个未来时间都是放在all变量中,all变量经过.sort()后会自动排序,从小到大。所以用时间戳而不是具体日期格式来计算出最近节点是可靠并简单的。

所以我们返回这个all[0]为最终输出。

结果如下:

image.png

那么这个时间到底对不对呢?

看一下我限制写这篇文章的时间:是6.5号深夜了

image.png

然后我们把这个时间戳翻译成日期:(见证奇迹的时刻来了!!!)

image.png


计算出的最近触发时间节点是:6.6号的早上6:30

完全正确!非常nice!


好,到此算是初步完成了领导的要求了。

虽然领导表示,你这个算法貌似不够完善,也不够健壮,更不符合pep8风格标准,甚至可能还存在着边界值bug。

但是整个的这个思路设计,非常好啊!非常值得组内推广!

于是,上传成了pypi的一个工具:python_jenkins_monitor

以后组员下载方便了,直接命令行:pip3 install python_jenkins_monitor 即可。

image.png

image.png

虽然后续该代码被领导给优化过了(加了好多标点符号和注释既视感),但是狗领导不给更新到pypi了。

大家如果想用的,自行优化下,或者留言讨论下吧~



相关文章
|
6月前
|
存储 JavaScript Java
(Python基础)新时代语言!一起学习Python吧!(四):dict字典和set类型;切片类型、列表生成式;map和reduce迭代器;filter过滤函数、sorted排序函数;lambda函数
dict字典 Python内置了字典:dict的支持,dict全称dictionary,在其他语言中也称为map,使用键-值(key-value)存储,具有极快的查找速度。 我们可以通过声明JS对象一样的方式声明dict
392 1
|
6月前
|
算法 Java Docker
(Python基础)新时代语言!一起学习Python吧!(三):IF条件判断和match匹配;Python中的循环:for...in、while循环;循环操作关键字;Python函数使用方法
IF 条件判断 使用if语句,对条件进行判断 true则执行代码块缩进语句 false则不执行代码块缩进语句,如果有else 或 elif 则进入相应的规则中执行
967 1
|
6月前
|
Java 数据处理 索引
(numpy)Python做数据处理必备框架!(二):ndarray切片的使用与运算;常见的ndarray函数:平方根、正余弦、自然对数、指数、幂等运算;统计函数:方差、均值、极差;比较函数...
ndarray切片 索引从0开始 索引/切片类型 描述/用法 基本索引 通过整数索引直接访问元素。 行/列切片 使用冒号:切片语法选择行或列的子集 连续切片 从起始索引到结束索引按步长切片 使用slice函数 通过slice(start,stop,strp)定义切片规则 布尔索引 通过布尔条件筛选满足条件的元素。支持逻辑运算符 &、|。
331 0
|
7月前
|
设计模式 缓存 监控
Python装饰器:优雅增强函数功能
Python装饰器:优雅增强函数功能
345 101
|
7月前
|
缓存 测试技术 Python
Python装饰器:优雅地增强函数功能
Python装饰器:优雅地增强函数功能
282 99
|
7月前
|
存储 缓存 测试技术
Python装饰器:优雅地增强函数功能
Python装饰器:优雅地增强函数功能
390 98
|
7月前
|
缓存 Python
Python中的装饰器:优雅地增强函数功能
Python中的装饰器:优雅地增强函数功能
|
8月前
|
Python
Python 函数定义
Python 函数定义
743 155
|
9月前
|
PHP Python
Python format()函数高级字符串格式化详解
在 Python 中,字符串格式化是一个重要的主题,format() 函数作为一种灵活且强大的字符串格式化方法,被广泛应用。format() 函数不仅能实现基本的插入变量,还支持更多高级的格式化功能,包括数字格式、对齐、填充、日期时间格式、嵌套字段等。 今天我们将深入解析 format() 函数的高级用法,帮助你在实际编程中更高效地处理字符串格式化。
706 0
|
6月前
|
存储 Java 调度
Python定时任务实战:APScheduler从入门到精通
APScheduler是Python强大的定时任务框架,通过触发器、执行器、任务存储和调度器四大组件,灵活实现各类周期性任务。支持内存、数据库、Redis等持久化存储,适用于Web集成、数据抓取、邮件发送等场景,解决传统sleep循环的诸多缺陷,助力构建稳定可靠的自动化系统。(238字)
1094 1