python定时任务模块APScheduler
,这个框架是专门处理定时任务的,目前还没怎么研究。所以我用了 threading
模块的Timer
。百度一下好多都是写个数据库做持续集成的。我想我运行自动化代码用例,不用去连接redis这些数据库实时刷新token当程序执行完毕定时任务自动结束是否可以行。所以就有下面的代码。但是也出现了麻烦,如果自动化代码运行时间超过服务器token失效时间比如1800秒也就是30分钟。那么我写的下面这段代码才会有意义些。如果2分钟就结束了,那么我设置30分钟,那得等28分钟才会结束。这尼玛这样30分钟内执行就不用实时更新token了。那岂不是我做了无用功。其实可以用jenkins做持续集成,没隔30分钟去执行一下你写的更新token的代码就行了。 但还是觉得不好,等以后遇到场景了再想办法吧。
from scripts.handle_request import to_request # 封装的requests请求
from scripts.handle_config import do_config # 封装读取配置文件
import threading
# 初始化配置文件的线程数,默认[]用来接收线程数的变化
def outer(func):
do_config.update_data("threading", "count", [])
def wrapper(*args,**kwargs):
res = func(*args, **kwargs)
return res
return wrapper
@outer
def get_token(url=None,data=None):
if url is None and data is None:
url = "http://127.0.0.1:8082/erp/loginIn"
data = {"name": "adeng35","pwd": "1111111111"}
res = to_request("POST", url=url,data=data, is_json=True)
global token
token = res.json()["token"]
print("当前token值:",token)
# 更新token
do_config.update_data("token","token",token)
print("---开始执行定时任务----")
a = threading.active_count()
b: list = do_config.get_eval_data("threading", "count")
b.append(a)
bb = len(b)
if bb > len(set(b)): # 线程数无变化时结束掉定时任务
return
else:
# 更新配置文件token值
do_config.update_data("threading", "count", b)
# 启动定时任务 30分钟执行一次,这里我设置6秒方便调试查看
threading.Timer(6,get_token).start()
get_token()
网友评论