python异步实现定时任务和周期任务的方法

 更新时间:2019年06月29日 10:14:45   作者:狡猾的皮球  
今天小编就为大家分享一篇python异步实现定时任务和周期任务的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

一. 如何调用

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异步实现定时任务和周期任务的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持脚本之家。

相关文章

  • django 实现celery动态设置周期任务执行时间

    django 实现celery动态设置周期任务执行时间

    今天小编就为大家分享一篇django 实现celery动态设置周期任务执行时间,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-11-11
  • PyChon中关于Jekins的详细安装(推荐)

    PyChon中关于Jekins的详细安装(推荐)

    这篇文章主要介绍了PyChon中关于Jekins的详细安装(推荐),本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-12-12
  • 详解python 降级到3.6终极解决方案

    详解python 降级到3.6终极解决方案

    这篇文章主要介绍了详解python 降级到3.6终极解决方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2020-02-02
  • Python的ORM框架SQLAlchemy入门教程

    Python的ORM框架SQLAlchemy入门教程

    SQLAlchemy 是python 操作数据库的一个库。能够进行 orm 映射,SQLAlchemy“采用简单的Python语言,为高效和高性能的数据库访问设计,实现了完整的企业级持久模型”
    2014-04-04
  • python实现三壶谜题的示例详解

    python实现三壶谜题的示例详解

    这篇文章主要介绍了python实现三壶谜题功能,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-11-11
  • Python-OpenCV:cv2.imread(),cv2.imshow(),cv2.imwrite()的区别

    Python-OpenCV:cv2.imread(),cv2.imshow(),cv2.imwrite()的区别

    本文主要介绍了OpenCV cv2.imread(),cv2.imshow(),cv2.imwrite()的区别,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2021-06-06
  • 使用Python将Mysql的查询数据导出到文件的方法

    使用Python将Mysql的查询数据导出到文件的方法

    今天小编就为大家分享一篇关于使用Python将Mysql的查询数据导出到文件的方法,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧
    2019-02-02
  • Django 日志配置按日期滚动的方法

    Django 日志配置按日期滚动的方法

    今天小编就为大家分享一篇Django 日志配置按日期滚动的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-01-01
  • Python从零开始创建区块链

    Python从零开始创建区块链

    这篇文章主要为大家详细介绍了Python从零开始创建区块链的步骤 ,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-03-03
  • pandas dataframe添加表格框线输出的方法

    pandas dataframe添加表格框线输出的方法

    今天小编就为大家分享一篇pandas dataframe添加表格框线输出的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-02-02

最新评论