如何用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
相关产品推荐
相关产品推荐

