我有一个可能执行时间很长的程序。在主模块中，我有以下内容：
import signal
def run_program()
...time consuming execution...
def Exit_gracefully(signal, frame):
... log exiting information ...
... close any open files ...
sys.exit(0)
if __name__ == '__main__':
signal.signal(signal.SIGINT, Exit_gracefully)
run_program()
这工作正常，但我希望有可能在捕获 SIGINT 时暂停执行，提示用户是否真的想退出，如果他们决定不想退出，则恢复我在 run_program() 中停止的地方。
我能想到的唯一方法是在一个单独的线程中运行程序，让主线程等待它并准备好捕获 SIGINT。如果用户想退出主线程，可以进行清理并杀死子线程。
有没有更简单的方法？
python 信号处理程序似乎不是真正的信号处理程序；也就是说，它们发生在事实之后，在正常流程中以及 C 处理程序已经返回之后。因此，您会尝试将退出逻辑放在信号处理程序中。由于信号处理程序在主线程中运行，它也会在那里阻塞执行。
像这样的东西似乎工作得很好。
import signal
import time
import sys
def run_program():
while True:
time.sleep(1)
print("a")
def exit_gracefully(signum, frame):
# restore the original signal handler as otherwise evil things will happen
# in raw_input when CTRL+C is pressed, and our signal handler is not re-entrant
signal.signal(signal.SIGINT, original_sigint)
try:
if raw_input("\nReally quit? (y/n)> ").lower().startswith('y'):
sys.exit(1)
except KeyboardInterrupt:
print("Ok ok, quitting")
sys.exit(1)
# restore the exit gracefully handler here
signal.signal(signal.SIGINT, exit_gracefully)
if __name__ == '__main__':
# store the original SIGINT handler
original_sigint = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, exit_gracefully)
run_program()
raw_input该代码在;的持续时间内恢复原始信号处理程序。raw_input它本身是不可重新进入的，重新进入它会导致RuntimeError: can't re-enter readline被提升，time.sleep这是我们不想要的，因为它比KeyboardInterrupt. 相反，我们让 2 个连续的 Ctrl-C 来 raise KeyboardInterrupt。
来自https://gist.github.com/rtfpessoa/e3b1fe0bbfcd8ac853bf
#!/usr/bin/env python
import signal
import sys
def signal_handler(signal, frame):
# your code here
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
再见！
假设你只想让程序在任务结束后做点什么
import time
class TestTask:
def __init__(self, msg: str):
self.msg = msg
def __enter__(self):
print(f'Task Start!:{self.msg}')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print('Task End!')
@staticmethod
def do_something():
try:
time.sleep(5)
except:
pass
with TestTask('Hello World') as task:
task.do_something()
当进程离开时，即使使用 KeyboardInterruptwith也会运行，这是相同的。__exit__
如果您不喜欢看到错误，请添加try ... except ...
@staticmethod
def do_something():
try:
time.sleep(5)
except:
pass
我没有完美的解决方案，但它可能对你有用。
这意味着将您的流程划分为许多子流程并保存完成。它不会再次执行，因为您发现它已经完成。
import time
from enum import Enum
class Action(Enum):
EXIT = 0
CONTINUE = 1
RESET = 2
class TestTask:
def __init__(self, msg: str):
self.msg = msg
def __enter__(self):
print(f'Task Start!:{self.msg}')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print('Task End!')
def do_something(self):
tuple_job = (self._foo, self._bar) # implement by yourself
list_job_state = [0] * len(tuple_job)
dict_keep = {} # If there is a need to communicate between jobs, and you don’t want to use class members, you can use this method.
while 1:
try:
for idx, cur_process in enumerate(tuple_job):
if not list_job_state[idx]:
cur_process(dict_keep)
list_job_state[idx] = True
if all(list_job_state):
print('100%')
break
except KeyboardInterrupt:
print('KeyboardInterrupt. input action:')
msg = '\n\t'.join([f"{action + ':':<10}{str(act_number)}" for act_number, action in
enumerate([name for name in vars(Action) if not name.startswith('_')])
])
case = Action(int(input(f'\t{msg}\n:')))
if case == Action.EXIT:
break
if case == Action.RESET:
list_job_state = [0] * len(tuple_job)
@staticmethod
def _foo(keep_dict: dict) -> bool: # implement by yourself
time.sleep(2)
print('1%')
print('2%')
print('...')
print('60%')
keep_dict['status_1'] = 'status_1'
return True
@staticmethod
def _bar(keep_dict: dict) -> bool: # implement by yourself
time.sleep(2)
print('61%')
print(keep_dict.get('status_1'))
print('...')
print('99%')
return True
with TestTask('Hello World') as task:
task.do_something()
安慰
input action number:2
Task Start!:Hello World
1%
2%
...
60%
KeyboardInterrupt. input action:
EXIT: 0
CONTINUE: 1
RESET: 2
:1
61%
status_1
...
99%
100%
Task End!