神剑山庄资源网 Design By www.hcban.com
一. 如何调用
def f1(arg1, arg2): print('f1', arg1, arg2) def f2(arg1): print('f2', arg1) def f3(): print('f3') def f4(): print('周期任务', int(time.time())) timer = TaskTimer() # 把任务加入任务队列 timer.join_task(f1, [1, 2], timing=15.5) # 每天15:30执行 timer.join_task(f2, [3], timing=14) # 每天14:00执行 timer.join_task(f3, [], timing=15) # 每天15:00执行 timer.join_task(f4, [], interval=10) # 每10秒执行1次 # 开始执行(此时才会创建线程) timer.start()
f1~f4是我们需要定时执行的函数。
首先创建TaskTimer对象(TaskTimer的代码在下面)。调用join_task函数,把需要执行的函数加入到任务队列。最后调用start,任务就开始执行了。
join_task参数:
fun:需要执行的函数
arg:fun的参数,如果没有就传一个空列表
interval:如果有此参数,说明任务是周期任务,单位为秒(注意interval最少5秒)
timing:如果有此参数,说明任务是定时任务,单位为时
注意:interval和timing只能选填1个
二. 源码
import datetime import time from threading import Thread from time import sleep class TaskTimer: __instance = None def __new__(cls, *args, **kwargs): """ 单例模式 """ if not cls.__instance: cls.__instance = object.__new__(cls) return cls.__instance def __init__(self): if not hasattr(self, 'task_queue'): setattr(self, 'task_queue', []) if not hasattr(self, 'is_running'): setattr(self, 'is_running', False) def write_log(self, level, msg): cur_time = datetime.datetime.now() with open('./task.log', mode='a+', encoding='utf8') as file: s = "[" + str(cur_time) + "][" + level + "] " + msg print(s) file.write(s + "\n") def work(self): """ 处理任务队列 """ while True: for task in self.task_queue: if task['interval']: self.cycle_task(task) elif task['timing']: self.timing_task(task) sleep(5) def cycle_task(self, task): """ 周期任务 """ if task['next_sec'] <= int(time.time()): try: task['fun'](*task['arg']) self.write_log("正常", "周期任务:" + task['fun'].__name__ + " 已执行") except Exception as e: self.write_log("异常", "周期任务:" + task['fun'].__name__ + " 函数内部异常:" + str(e)) finally: task['next_sec'] = int(time.time()) + task['interval'] def timing_task(self, task): """ 定时任务 """ # 今天已过秒数 today_sec = self.get_today_until_now() # 到了第二天,就重置任务状态 if task['today'] != self.get_today(): task['today'] = self.get_today() task['today_done'] = False # 第一次执行 if task['first_work']: if today_sec >= task['task_sec']: task['today_done'] = True task['first_work'] = False else: task['first_work'] = False # 今天还没有执行 if not task['today_done']: if today_sec >= task['task_sec']: # 到点了,开始执行任务 try: task['fun'](*task['arg']) self.write_log("正常", "定时任务:" + task['fun'].__name__ + " 已执行") except Exception as e: self.write_log("异常", "定时任务:" + task['fun'].__name__ + " 函数内部异常:" + str(e)) finally: task['today_done'] = True if task['first_work']: task['first_work'] = False def get_today_until_now(self): """ 获取今天凌晨到现在的秒数 """ i = datetime.datetime.now() return i.hour * 3600 + i.minute * 60 + i.second def get_today(self): """ 获取今天的日期 """ i = datetime.datetime.now() return i.day def join_task(self, fun, arg, interval=None, timing=None): """ interval和timing只能存在1个 :param fun: 你要调用的任务 :param arg: fun的参数 :param interval: 周期任务,单位秒 :param timing: 定时任务,取值:[0,24) """ # 参数校验 if (interval != None and timing != None) or (interval == None and timing == None): raise Exception('interval和timing只能选填1个') if timing and not 0 <= timing < 24: raise Exception('timing的取值范围为[0,24)') if interval and interval < 5: raise Exception('interval最少为5') # 封装一个task task = { 'fun': fun, 'arg': arg, 'interval': interval, 'timing': timing, } # 封装周期或定时任务相应的参数 if timing: task['task_sec'] = timing * 3600 task['today_done'] = False task['first_work'] = True task['today'] = self.get_today() elif interval: task['next_sec'] = int(time.time()) + interval # 把task加入任务队列 self.task_queue.append(task) self.write_log("正常", "新增任务:" + fun.__name__) def start(self): """ 开始执行任务 返回线程标识符 """ if not self.is_running: thread = Thread(target=self.work) thread.start() self.is_running = True self.write_log("正常", "TaskTimer已开始运行!") return thread.ident self.write_log("警告", "TaskTimer已运行,请勿重复启动!")
以上这篇python异步实现定时任务和周期任务的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持。
神剑山庄资源网 Design By www.hcban.com
神剑山庄资源网
免责声明:本站文章均来自网站采集或用户投稿,网站不提供任何软件下载或自行开发的软件!
如有用户或公司发现本站内容信息存在侵权行为,请邮件告知! 858582#qq.com
神剑山庄资源网 Design By www.hcban.com
暂无python异步实现定时任务和周期任务的方法的评论...
RTX 5090要首发 性能要翻倍!三星展示GDDR7显存
三星在GTC上展示了专为下一代游戏GPU设计的GDDR7内存。
首次推出的GDDR7内存模块密度为16GB,每个模块容量为2GB。其速度预设为32 Gbps(PAM3),但也可以降至28 Gbps,以提高产量和初始阶段的整体性能和成本效益。
据三星表示,GDDR7内存的能效将提高20%,同时工作电压仅为1.1V,低于标准的1.2V。通过采用更新的封装材料和优化的电路设计,使得在高速运行时的发热量降低,GDDR7的热阻比GDDR6降低了70%。
更新日志
2024年10月05日
2024年10月05日
- 群星《前途海量 电影原声专辑》[FLAC/分轨][227.78MB]
- 张信哲.1992-知道新曲与精丫巨石】【WAV+CUE】
- 王翠玲.1995-ANGEL【新艺宝】【WAV+CUE】
- 景冈山.1996-我的眼里只有你【大地唱片】【WAV+CUE】
- 群星《八戒 电影原声带》[320K/MP3][188.97MB]
- 群星《我的阿勒泰 影视原声带》[320K/MP3][139.47MB]
- 纪钧瀚《胎教古典音乐 钢琴与大提琴的沉浸时光》[320K/MP3][148.91MB]
- 刘雅丽.2001-丽花皇后·EMI精选王【EMI百代】【FLAC分轨】
- 齐秦.1994-黄金十年1981-1990CHINA.TOUR.LIVE精丫上华】【WAV+CUE】
- 群星.2008-本色·百代音乐人创作专辑【EMI百代】【WAV+CUE】
- 群星.2001-同步过冬AVCD【环球】【WAV+CUE】
- 群星.2020-同步过冬2020冀待晴空【环球】【WAV+CUE】
- 沈雁.1986-四季(2012梦田复刻版)【白云唱片】【WAV+CUE】
- 纪钧瀚《胎教古典音乐 钢琴与大提琴的沉浸时光》[FLAC/分轨][257.88MB]
- 《国语老歌 怀旧篇 3CD》[WAV/分轨][1.6GB]