【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
相关文章
|
5天前
|
缓存 监控 测试技术
Python中的装饰器:功能扩展与代码复用的利器###
本文深入探讨了Python中装饰器的概念、实现机制及其在实际开发中的应用价值。通过生动的实例和详尽的解释,文章展示了装饰器如何增强函数功能、提升代码可读性和维护性,并鼓励读者在项目中灵活运用这一强大的语言特性。 ###
|
8天前
|
缓存 开发者 Python
探索Python中的装饰器:简化代码,增强功能
【10月更文挑战第35天】装饰器在Python中是一种强大的工具,它允许开发者在不修改原有函数代码的情况下增加额外的功能。本文旨在通过简明的语言和实际的编码示例,带领读者理解装饰器的概念、用法及其在实际编程场景中的应用,从而提升代码的可读性和复用性。
|
4天前
|
Python
探索Python中的装饰器:简化代码,提升效率
【10月更文挑战第39天】在编程的世界中,我们总是在寻找使代码更简洁、更高效的方法。Python的装饰器提供了一种强大的工具,能够让我们做到这一点。本文将深入探讨装饰器的基本概念,展示如何通过它们来增强函数的功能,同时保持代码的整洁性。我们将从基础开始,逐步深入到装饰器的高级用法,让你了解如何利用这一特性来优化你的Python代码。准备好让你的代码变得更加优雅和强大了吗?让我们开始吧!
12 1
|
9天前
|
设计模式 缓存 监控
Python中的装饰器:代码的魔法增强剂
在Python编程中,装饰器是一种强大而灵活的工具,它允许程序员在不修改函数或方法源代码的情况下增加额外的功能。本文将探讨装饰器的定义、工作原理以及如何通过自定义和标准库中的装饰器来优化代码结构和提高开发效率。通过实例演示,我们将深入了解装饰器的应用,包括日志记录、性能测量、事务处理等常见场景。此外,我们还将讨论装饰器的高级用法,如带参数的装饰器和类装饰器,为读者提供全面的装饰器使用指南。
|
5天前
|
存储 缓存 监控
掌握Python装饰器:提升代码复用性与可读性的利器
在本文中,我们将深入探讨Python装饰器的概念、工作原理以及如何有效地应用它们来增强代码的可读性和复用性。不同于传统的函数调用,装饰器提供了一种优雅的方式来修改或扩展函数的行为,而无需直接修改原始函数代码。通过实际示例和应用场景分析,本文旨在帮助读者理解装饰器的实用性,并鼓励在日常编程实践中灵活运用这一强大特性。
|
9天前
|
存储 算法 搜索推荐
Python高手必备!揭秘图(Graph)的N种风骚表示法,让你的代码瞬间高大上
在Python中,图作为重要的数据结构,广泛应用于社交网络分析、路径查找等领域。本文介绍四种图的表示方法:邻接矩阵、邻接表、边列表和邻接集。每种方法都有其特点和适用场景,掌握它们能提升代码效率和可读性,让你在项目中脱颖而出。
22 5
|
7天前
|
机器学习/深度学习 数据采集 人工智能
探索机器学习:从理论到Python代码实践
【10月更文挑战第36天】本文将深入浅出地介绍机器学习的基本概念、主要算法及其在Python中的实现。我们将通过实际案例,展示如何使用scikit-learn库进行数据预处理、模型选择和参数调优。无论你是初学者还是有一定基础的开发者,都能从中获得启发和实践指导。
17 2
|
9天前
|
数据库 Python
异步编程不再难!Python asyncio库实战,让你的代码流畅如丝!
在编程中,随着应用复杂度的提升,对并发和异步处理的需求日益增长。Python的asyncio库通过async和await关键字,简化了异步编程,使其变得流畅高效。本文将通过实战示例,介绍异步编程的基本概念、如何使用asyncio编写异步代码以及处理多个异步任务的方法,帮助你掌握异步编程技巧,提高代码性能。
26 4
|
11天前
|
缓存 开发者 Python
探索Python中的装饰器:简化和增强你的代码
【10月更文挑战第32天】 在编程的世界中,简洁和效率是永恒的追求。Python提供了一种强大工具——装饰器,它允许我们以声明式的方式修改函数的行为。本文将深入探讨装饰器的概念、用法及其在实际应用中的优势。通过实际代码示例,我们不仅理解装饰器的工作方式,还能学会如何自定义装饰器来满足特定需求。无论你是初学者还是有经验的开发者,这篇文章都将为你揭示装饰器的神秘面纱,并展示如何利用它们简化和增强你的代码库。
|
9天前
|
API 数据处理 Python
探秘Python并发新世界:asyncio库,让你的代码并发更优雅!
在Python编程中,随着网络应用和数据处理需求的增长,并发编程变得愈发重要。asyncio库作为Python 3.4及以上版本的标准库,以其简洁的API和强大的异步编程能力,成为提升性能和优化资源利用的关键工具。本文介绍了asyncio的基本概念、异步函数的定义与使用、并发控制和资源管理等核心功能,通过具体示例展示了如何高效地编写并发代码。
20 2