Python中如何为函数设置倒计时定时器实现回调触发逻辑

这个需求完全可以实现，核心是通过定时器实现2分钟的回调缓冲窗口：窗口开启阶段只暂存回调传入的参数不执行业务逻辑，窗口到期后统一处理所有收集到的任务，处理完成后自动复位等待下一轮触发即可。

• 修正原代码的结构问题：去掉全局变量、避免每次触发回调都新建类实例，将待处理队列、调用计数器、定时器对象都作为类实例属性统一维护
• 首次触发initialize回调时，启动时长120秒的倒计时定时器；2分钟窗口内所有新触发的回调只做URL入队、计数累加操作，不执行业务处理
• 2分钟倒计时结束后，先复位定时器标记，再依次执行预设操作、批量跑main方法处理窗口内收集到的所有URL
• 所有任务处理完成后，清空暂存队列、重置计数器，回到等待剪贴板新信号的状态，下次收到触发信号时自动启动新的2分钟倒计时

import clipboard_monitor
from threading import Timer

class ABC():
    def __init__(self):
        self.to_do = []  # 待处理URL队列
        self.countdown_timer = None  # 2分钟倒计时定时器实例
        self.call_count = 0  # 当前窗口内initialize被调用次数
        self.WINDOW_DURATION = 120  # 缓冲窗口时长，单位：秒

    def initialize(self, url):
        # 新URL入队，累加调用计数
        self.to_do.append(url)
        self.call_count += 1
        print(f"URL has been added to queue for download. Current queue count: {self.call_count}")

        # 无运行中定时器时，启动新的2分钟倒计时
        if self.countdown_timer is None:
            self.countdown_timer = Timer(self.WINDOW_DURATION, self._process_pending_tasks)
            self.countdown_timer.start()

    def _process_pending_tasks(self):
        # 定时器到期先复位标记，方便下一轮触发新建定时器
        self.countdown_timer = None
        # 此处可插入你需要的其他预设操作
        print("2 minutes window reached, start processing tasks...")
        # 批量处理窗口内收集的所有URL
        for url in self.to_do:
            self.main(url)
        # 处理完成后重置状态，等待下一轮触发
        self.to_do.clear()
        self.call_count = 0

    def main(self, url):
        # 替换为你原有的main业务逻辑
        print(f"Processing url: {url}")

if __name__ == "__main__":
    app = ABC()
    # 绑定实例方法作为剪贴板监听回调
    clipboard_monitor.on_text(app.initialize)
    clipboard_monitor.wait()

• 如果使用的剪贴板监听在子线程触发回调，担心队列操作出现竞态，可以给入队、批量处理逻辑加线程锁，普通单线程监听场景下上述代码可稳定运行
• 直接修改WINDOW_DURATION参数即可自由调整缓冲窗口的时长
• 如需在窗口到期后新增其他前置操作，直接在_process_pending_tasks方法中、遍历执行main逻辑前插入对应代码即可

内容的提问来源于stack exchange，提问作者Zuss