我正在编写一个启动多个进程和数据库连接的 python 脚本。我时不时地想用 Ctrl+C 信号终止脚本,并且我想做一些清理工作。
在 Perl 中我会这样做:
$SIG{'INT'} = 'exit_gracefully';
sub exit_gracefully {
print "Caught ^C \n";
exit (0);
}
如何在 Python 中进行类似操作?
signal.signal
注册您的处理程序,如下所示:
#!/usr/bin/env python
import signal
import sys
def signal_handler(sig, frame):
print('You pressed Ctrl+C!')
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
print('Press Ctrl+C')
signal.pause()
代码改编自这里。
有关
signal
的更多文档可以在 here 找到。
您可以将其视为异常 (
KeyboardInterrupt
),就像任何其他异常一样。创建一个新文件并从 shell 中运行它,其中包含以下内容,看看我的意思:
import time, sys
x = 1
while True:
try:
print x
time.sleep(.3)
x += 1
except KeyboardInterrupt:
print "Bye"
sys.exit()
作为上下文管理器:
import signal
class GracefulInterruptHandler(object):
def __init__(self, sig=signal.SIGINT):
self.sig = sig
def __enter__(self):
self.interrupted = False
self.released = False
self.original_handler = signal.getsignal(self.sig)
def handler(signum, frame):
self.release()
self.interrupted = True
signal.signal(self.sig, handler)
return self
def __exit__(self, type, value, tb):
self.release()
def release(self):
if self.released:
return False
signal.signal(self.sig, self.original_handler)
self.released = True
return True
使用方法:
with GracefulInterruptHandler() as h:
for i in xrange(1000):
print "..."
time.sleep(1)
if h.interrupted:
print "interrupted!"
time.sleep(2)
break
嵌套处理程序:
with GracefulInterruptHandler() as h1:
while True:
print "(1)..."
time.sleep(1)
with GracefulInterruptHandler() as h2:
while True:
print "\t(2)..."
time.sleep(1)
if h2.interrupted:
print "\t(2) interrupted!"
time.sleep(2)
break
if h1.interrupted:
print "(1) interrupted!"
time.sleep(2)
break
您可以通过捕获 KeyboardInterrupt
异常来处理 CTRL+
C。您可以在异常处理程序中实现任何清理代码。
将
main
称为主函数,将 exit_gracefully
称为 Ctrl+C 处理程序
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
pass
finally:
exit_gracefully()
来自Python的文档:
import signal
import time
def handler(signum, frame):
print 'Here you go'
signal.signal(signal.SIGINT, handler)
time.sleep(10) # Press Ctrl+c here
我改编了@udi 的代码以支持多个信号(没什么花哨的):
class GracefulInterruptHandler(object):
def __init__(self, signals=(signal.SIGINT, signal.SIGTERM)):
self.signals = signals
self.original_handlers = {}
def __enter__(self):
self.interrupted = False
self.released = False
for sig in self.signals:
self.original_handlers[sig] = signal.getsignal(sig)
signal.signal(sig, self.handler)
return self
def handler(self, signum, frame):
self.release()
self.interrupted = True
def __exit__(self, type, value, tb):
self.release()
def release(self):
if self.released:
return False
for sig in self.signals:
signal.signal(sig, self.original_handlers[sig])
self.released = True
return True
此代码支持键盘中断调用(
SIGINT
)和SIGTERM
(kill <process>
)
如果您想确保清理过程完成,我会使用 SIG_IGN 添加到 Matt J 的答案,这样进一步的
SIGINT
就会被忽略,这将防止您的清理被中断。
import signal
import sys
def signal_handler(signum, frame):
signal.signal(signum, signal.SIG_IGN) # ignore additional signals
cleanup() # give your process a chance to clean up
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler) # register the signal with the signal handler first
do_stuff()
与 Matt J 他的答案相反,我使用一个简单的对象。这使我可以将此处理程序解析为所有需要停止安全的线程。
class SIGINT_handler():
def __init__(self):
self.SIGINT = False
def signal_handler(self, signal, frame):
print('You pressed Ctrl+C!')
self.SIGINT = True
handler = SIGINT_handler()
signal.signal(signal.SIGINT, handler.signal_handler)
其他地方
while True:
# task
if handler.SIGINT:
break
您可以使用Python内置的信号模块中的函数在Python中设置信号处理程序。具体来说,
signal.signal(signalnum, handler)
函数用于注册信号handler
的signalnum
函数。
感谢现有的答案,但添加了
signal.getsignal()
import signal
# store default handler of signal.SIGINT
default_handler = signal.getsignal(signal.SIGINT)
catch_count = 0
def handler(signum, frame):
global default_handler, catch_count
catch_count += 1
print ('wait:', catch_count)
if catch_count > 3:
# recover handler for signal.SIGINT
signal.signal(signal.SIGINT, default_handler)
print('expecting KeyboardInterrupt')
signal.signal(signal.SIGINT, handler)
print('Press Ctrl+c here')
while True:
pass
就我个人而言,我无法使用 try/ except KeyboardInterrupt,因为我使用的是阻塞的标准套接字(IPC)模式。因此,SIGINT 已被提示,但仅在套接字上接收到数据后才出现。
设置信号处理程序的行为相同。
另一方面,这仅适用于实际终端。其他启动环境可能不接受 Ctrl+C,或预先处理信号。
此外,Python 中还有“Exceptions”和“BaseExceptions”,它们的区别在于解释器需要自行干净退出,因此某些异常比其他异常具有更高的优先级(Exceptions 源自 BaseException)
这是一个单行代码(减去导入),使用 Python 3.10 进行测试(不确定其他版本):
#!/usr/bin/env python3
from signal import SIGINT, SIGTERM, sigwait
print('received', sigwait({SIGINT, SIGTERM}))
尽管如果您在 INT/TERM 之前发送另一个信号(例如 USR1),这种行为会很有趣。 sigwait 将等待 INT/TERM,但不会按预期输出“received {signal}”行。
$ ./test.py # ctrl + c
^Creceived Signals.SIGINT
$ ./test.py # kill -TERM <pid>
received Signals.SIGTERM
$ ./test.py # kill -USR1 <pid> ; kill -TERM <pid>
User defined signal 1: 30