【Python】300行代码实现crontab定时器功能 【下】

简介: 熟悉Linux的都知道在Linux下有一个crontab的定时任务,可以很方便的进行各种定时、计划任务的执行。有时候写代码也需要用到定时器业务,因此我使用Python实现了一个类似的定时器模块,可以很方便的做定时业务,使用例子如下
 # 只有满足条件才执行
     if self.call_condition is None or self.call_condition != self.cur_condition:
        self.call_condition = self.cur_condition
        return True
     return False
  def call(self):
     """执行定时回调"""
     try:
        self.call_able(*self.call_args)
     except Exception as e:
        print("call", self.call_able, "error:", e)
  @staticmethod
  def check_target_value(src_value, des_value):
     """检查指定值"""
     return src_value == des_value
  @staticmethod
  def check_delay_value(src_value, des_value):
     """检查每隔多久"""
        return src_value > des_value:
  @staticmethod
  def check_range_value(src_value, min_value, max_value):
     """检查区间"""
     return src_value >= min_value and src_value <= max_value
  @staticmethod
  def check_multi_value(src_value, values):
     """检查多个值"""
     return src_value in values
  @staticmethod
  def _sub_months(st, ed):
     return (ed.year - st.year)*12 + (ed.month - st.month)
  @staticmethod
  def _parse_part(value):
     """解析单个数据"""
     value = value.strip()
     if value == "*":
        return -1, -1
     # 指定时间一次
     if value.isdigit():
        return 0, int(value, 10)
     # */v 每隔多久一次
     fraction_idx = value.find("/")
     if fraction_idx >= 0 :
        denominator = value[fraction_idx + 1:]
        if not denominator.isdigit():
            return -1, -1
        if value.startswith("*"):
           return 1, int(denominator, 10)
        else:
           # 特殊区间A-B/N 每个多久,且在区间A,B内
           range_idx = value.find("-")
           if range_idx < 0:
              return -1, -1
           # 区间 + 间隔
           values = []
           range_values = value[:fraction_idx].split("-")
           if len(range_values) != 2:
              return -1, -1
           # 必须是数字,且只存在2个值
           if range_values[0].isdigit() and range_values[1].isdigit():
              values.append(int(range_values[0], 10))
              values.append(int(range_values[1], 10))
            values.append(int(denominator, 10))
            return 1, values
    # 多个时间点 A,B,...,Z
    dom_idx = value.find(",")
    if dom_idx >= 0:
       values = []
       for v in value.split(","):
           if v.isdigit():
               values.append(int(v, 10))
       return 2, values
     # 区间A-B
     range_idx = value.find("-")
     if range_idx >= 0:
         values = []
         range_values = value.split("-")
         if len(range_values) != 2:
            return -1, -1
         # 必须是数字,且只存在2个值
         if range_values[0].isdigit() and range_values[1].isdigit():
             values.append(int(range_values[0],10))
             values.append(int(range_values[1],10))
             return 3, values
         return -1, -1
    return -1, -1
class Timer:
    def __init__(self, not_thread=False):
        """创建定时器对象"""
        self._thread = None
        self._timer_data = {}
        if not not_thread:
            self._thread = threading.Thread(target=self._update, args=())
            self._thread.start()
    def register(self, time_fmt, call_able, *args):
        """注册定时器"""
        timer_data = TimerInfo(time_fmt, call_able, args)
        if not timer_data.enable:
            return
        timer_id = uuid.uuid4()
        self._timer_data[timer_id] = timer_data
        return timer_id
    def _update(self):
        """内部更新"""
        while self._thread:
            self.update(datetime.datetime.now())
            time.sleep(5)
    def update(self, date_time):
        """外部更新使用"""
        if not isinstance(date_time, datetime.datetime):
            raise TypeError("date_time must be datetime")
        remove_list = []
        for tid, node in self._timer_data.items():
            if not node.enable:
                remove_list.append(tid)
                continue
            if node.on_timer(date_time):
                node.call()
        # 删除已标记删除的定时器列表
        for tid in remove_list:
            del self._timer_data[tid]
    def remove(self, timer_id):
        """移除定时器"""
        if timer_id in self._timer_data:
            # 只标记
            self._timer_data[timer_id].enable = False
相关文章
|
8天前
|
开发者 Python
探索Python中的装饰器:简化代码,增强功能
【10月更文挑战第22天】在Python的世界里,装饰器是一个强大的工具,它能够让我们以简洁的方式修改函数的行为,增加额外的功能而不需要重写原有代码。本文将带你了解装饰器的基本概念,并通过实例展示如何一步步构建自己的装饰器,从而让你的代码更加高效、易于维护。
|
5天前
|
算法 测试技术 开发者
在Python开发中,性能优化和代码审查至关重要。性能优化通过改进代码结构和算法提高程序运行速度,减少资源消耗
在Python开发中,性能优化和代码审查至关重要。性能优化通过改进代码结构和算法提高程序运行速度,减少资源消耗;代码审查通过检查源代码发现潜在问题,提高代码质量和团队协作效率。本文介绍了一些实用的技巧和工具,帮助开发者提升开发效率。
10 3
|
9天前
|
开发框架 Python
探索Python中的装饰器:简化代码,增强功能
【10月更文挑战第20天】在编程的海洋中,简洁与强大是航行的双桨。Python的装饰器,这一高级特性,恰似海风助力,让代码更优雅、功能更强大。本文将带你领略装饰器的奥秘,从基础概念到实际应用,一步步深入其内涵与意义。
|
7天前
|
机器学习/深度学习 缓存 数据挖掘
Python性能优化:提升你的代码效率
【10月更文挑战第22天】 Python性能优化:提升你的代码效率
9 1
|
14天前
|
云计算 Python
用python给你写个简单的计算器功能网页啊
这张图片展示了阿里巴巴集团的组织架构图,涵盖了核心电商、云计算、数字媒体与娱乐、创新业务等主要板块,以及各板块下的具体业务单元和部门。
|
10天前
|
机器人 Shell Linux
【Azure Bot Service】部署Python ChatBot代码到App Service中
本文介绍了使用Python编写的ChatBot在部署到Azure App Service时遇到的问题及解决方案。主要问题是应用启动失败,错误信息为“Failed to find attribute &#39;app&#39; in &#39;app&#39;”。解决步骤包括:1) 修改`app.py`文件,添加`init_func`函数;2) 配置`config.py`,添加与Azure Bot Service认证相关的配置项;3) 设置App Service的启动命令为`python3 -m aiohttp.web -H 0.0.0.0 -P 8000 app:init_func`。
|
14天前
|
人工智能 IDE 测试技术
使用通义灵码提升Python开发效率:从熟悉代码到实现需求的全流程体验
作为一名Python开发者,我最近开始使用通义灵码作为开发辅助工具。它显著提高了我的工作效率,特别是在理解和修改复杂代码逻辑方面。通过AI编码助手,我能够在短时间内快速上手新项目,实现新需求,并进行代码优化,整体效率提升了60%以上。通义灵码不仅加快了代码生成速度,还增强了代码的健壮性和稳定性。
|
14天前
|
数据处理 开发者 Python
Python中的列表推导式:一种优雅的代码简化技巧####
【10月更文挑战第15天】 本文将深入浅出地探讨Python中列表推导式的使用,这是一种强大且简洁的语法结构,用于从现有列表生成新列表。通过具体示例和对比传统循环方法,我们将揭示列表推导式如何提高代码的可读性和执行效率,同时保持语言的简洁性。无论你是Python初学者还是有经验的开发者,掌握这一技能都将使你的编程之旅更加顺畅。 ####
18 1
|
16天前
|
缓存 程序员 开发者
探索Python中的装饰器:一种优雅的代码增强技巧
【10月更文挑战第13天】 在本文中,我们将深入探讨Python中的装饰器,这是一种强大的工具,它允许程序员以简洁而高效的方式扩展或修改函数和类的行为。通过具体示例,我们将展示如何利用装饰器来优化代码结构,提高开发效率,并实现如日志记录、性能计时等常见功能。本文旨在为读者提供一个关于Python装饰器的全面理解,从而能够在他们的项目中灵活运用这一技术。
21 1
|
20天前
|
设计模式 开发者 Python
Python中的装饰器:简化代码与增强功能
【10月更文挑战第9天】在编程的世界里,效率和可读性是衡量代码质量的两大关键指标。Python语言以其简洁明了的语法赢得了无数开发者的青睐,而装饰器则是其独特魅力之一。本文将深入探讨装饰器的工作原理、使用方法以及如何通过自定义装饰器来提升代码的重用性和可维护性,让读者能够更加高效地编写出既优雅又功能强大的代码。