这是一个可能的解决方法:使用
time.sleep()
代替
Timer
意味着可以实现“优雅的关闭”机制...适用于Python3,因为在其中,似乎
KeyboardInterrupt
仅在主线程的用户代码中引发。否则,似乎该异常会被“忽略”,就像
here所述:实际上,它会导致出现异常的线程立即死亡,但不会影响任何祖先线程,这很棘手,因为无法捕获异常。
假设你希望Ctrl-C的响应时间为0.5秒,但你只想每5秒重复一些随机持续时间的实际工作:
import threading, sys, time, random
blip_counter = 0
work_threads=[]
def repeat_every_5():
global blip_counter
print( f'counter: {blip_counter}')
def real_work():
real_work_duration_s = random.randrange(10)
print( f'do some real work every 5 seconds, lasting {real_work_duration_s} s: starting...')
for interval_500ms in range( real_work_duration_s * 2 ):
if threading.current_thread().stop_event.is_set():
print( f'stop_event SET!')
return
time.sleep(0.5)
print( f'...real work ends')
for work_thread in work_threads:
if not work_thread.is_alive():
print(f'work thread {work_thread} dead, removing from list' )
work_threads.remove( work_thread )
new_work_thread = threading.Thread(target=real_work)
new_work_thread.stop_event = threading.Event()
work_threads.append(new_work_thread)
new_work_thread.start()
blip_counter += 1
time.sleep( 5 )
timer_thread = threading.Thread(target=repeat_every_5)
timer_thread.daemon = True
timer_thread.start()
repeat_every_5()
while True:
try:
time.sleep( 0.5 )
except KeyboardInterrupt:
print( f'shutting down due to Ctrl-C..., work threads left: {len(work_threads)}')
for work_thread in work_threads:
if work_thread.is_alive():
print( f'work_thread {work_thread}: setting STOP event')
work_thread.stop_event.set()
print( f'work_thread {work_thread}: joining to main...')
work_thread.join()
print( f'work_thread {work_thread}: ...joined to main')
else:
print( f'work_thread {work_thread} has died' )
sys.exit(1)
这个 while True:
机制看起来有点笨重。但我认为,目前(Python 3.8.x)KeyboardInterrupt
只能在主线程中捕获。
PS 根据我的实验,处理子进程可能更容易,因为在简单情况下,Ctrl-C 似乎会导致所有正在运行的进程同时发生 KeyboardInterrupt
。