You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用Python实现支持暂停、恢复、终止且带事件回调的倒计时器?

问题根源

你当前的代码是单线程同步执行逻辑,count.run() 是阻塞方法,会占用主线程跑完10秒倒计时后才会执行后续的time.sleep(3)和count.pause(),自然无法实现中途暂停的效果。同时要实现事件绑定机制,需要提前存储不同事件对应的回调函数,事件触发时遍历执行即可。

完整修改实现
import time
import threading

class Countdown:
    def __init__(self, total_time, finish_callback):
        self.total_time = total_time
        self.remain_time = total_time
        self.finish_callback = finish_callback
        # 状态标志
        self.paused = False
        self.aborted = False
        # 事件回调存储
        self.event_handlers = {}
        # 运行线程
        self._run_thread = None
        self._finish_args = ()

    def on(self, event_name, callback):
        """绑定事件回调,支持的事件:pause/resume/abort/finish"""
        if event_name not in self.event_handlers:
            self.event_handlers[event_name] = []
        self.event_handlers[event_name].append(callback)

    def _trigger_event(self, event_name, *args, **kwargs):
        """触发指定事件,执行所有绑定的回调"""
        for callback in self.event_handlers.get(event_name, []):
            callback(*args, **kwargs)

    def pause(self):
        """暂停倒计时"""
        if not self.paused and self.remain_time > 0 and not self.aborted:
            self.paused = True
            self._trigger_event("pause")

    def resume(self):
        """恢复倒计时"""
        if self.paused and self.remain_time > 0 and not self.aborted:
            self.paused = False
            self._trigger_event("resume")

    def abort(self):
        """终止倒计时"""
        if not self.aborted:
            self.aborted = True
            self._trigger_event("abort")
            if self._run_thread:
                self._run_thread.join()

    def _countdown_task(self):
        """倒计时核心逻辑,运行在独立子线程"""
        while self.remain_time > 0 and not self.aborted:
            # 等待暂停结束
            while self.paused and not self.aborted:
                time.sleep(0.1)
            if self.aborted:
                break
            print(self.remain_time)
            # 拆分为10次短睡眠,保证控制指令可以即时响应
            for _ in range(10):
                if self.paused or self.aborted:
                    break
                time.sleep(0.1)
            if not self.paused and not self.aborted:
                self.remain_time -= 1

        # 倒计时结束/终止处理
        if self.aborted:
            return
        self._trigger_event("finish")
        self.finish_callback(*self._finish_args)

    def start(self, *finish_args):
        """启动倒计时"""
        self._finish_args = finish_args
        self._run_thread = threading.Thread(target=self._countdown_task, daemon=True)
        self._run_thread.start()


# 测试用例
def sample_fun(Yname,Yage):
    print('hi {a} you are {b} years old.'.format(a=Yname,b=Yage))

def on_pause_handler():
    print("触发暂停事件,当前剩余时间:", count.remain_time)

def on_resume_handler():
    print("触发恢复事件,继续倒计时")

# 初始化倒计时
count = Countdown(10, sample_fun)
# 绑定事件回调
count.on("pause", on_pause_handler)
count.on("resume", on_resume_handler)
# 启动倒计时,传入结束回调的参数
count.start('John Doe','30')
# 主线程等待3秒后调用暂停
time.sleep(3)
count.pause()
# 暂停2秒后恢复
time.sleep(2)
count.resume()
# 如果要中途终止可以调用 count.abort()
# 等待倒计时结束
while count.remain_time >0 and not count.aborted:
    time.sleep(0.1)
核心改动说明
  • 使用Python标准库threading将倒计时逻辑放到独立子线程运行,不会阻塞主线程的控制指令调用,符合你“不引入外部库”的要求
  • 新增on()方法实现事件绑定机制,支持pause、resume、abort、finish四种事件的回调绑定
  • 把原来整1秒的睡眠拆分为10次0.1秒的短睡眠,保证暂停、终止指令可以在100ms内响应,不会等整秒结束才生效
  • 暂停状态使用短间隔循环检测,不会占用过高CPU资源
  • 所有状态变更都会自动触发对应事件的绑定回调,满足自定义事件处理需求

内容的提问来源于stack exchange,提问作者Tara Nath Niraula

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.08 13:45:01