在等待事件的程序中捕获键盘中断

Question

以下程序会挂起终端，使其忽略Ctrl+C。这非常令人烦恼，因为每当其中一个线程挂起时，我都必须重新启动终端。

在等待事件时是否有任何方法可以捕获KeyboardInterrupt？

import threading
def main():
    finished_event = threading.Event()
    startThread(finished_event)
    finished_event.wait()#I want to stop the program here
    print('done!')
def startThread(evt):
    """Start a thread that will trigger evt when it is done"""
    #evt.set()
if __name__ == '__main__':
    main()

- Navin

Answer 1

如果你想避免轮询，可以使用 signal 模块的 pause() 函数，而不是使用 finished_event.wait()。 signal.pause() 是一个阻塞函数，当进程接收到信号时就会解除阻塞。在这种情况下，当按下 ^C 键时，SIGINT 信号会解除函数的阻塞。请注意，根据文档，该函数在 Windows 上无法工作。我在 Linux 上尝试过，并且它对我有效。

我在这个 SO 帖子 中发现了这个解决方案。

- GingerNinja23

---

有点晚了，但感谢您提供了一个非轮询的答案 :) - Navin

我认为你可以编辑你的回答，使其更简单和直接，但是你解决了我的问题，谢谢！ - Almog Cohen

Answer 2

更新: 在当前Python 3版本中，finished_event.wait()在我的Ubuntu机器上可用（从Python 3.2开始）。您不需要指定timeout参数，只需使用Ctrl+C中断它即可。在CPython 2上需要传递timeout参数。

以下是完整的代码示例：

#!/usr/bin/env python3
import threading

def f(event):
    while True:
        pass
    # never reached, otherwise event.set() would be here

event = threading.Event()
threading.Thread(target=f, args=[event], daemon=True).start()
try:
    print('Press Ctrl+C to exit')
    event.wait()
except KeyboardInterrupt:
    print('got Ctrl+C')

与Ctrl+C相关的错误是有可能存在的。请在您的环境中测试它是否有效。

---

^{旧轮询答案：}

您可以尝试允许解释器运行主线程：

while not finished_event.wait(.1): # timeout in seconds
    pass

如果你只想等到子线程完成：

while thread.is_alive():
    thread.join(.1)

- jfs

---

我不知道。它是否适用于轮询？为什么轮询不好？您可以测试 time.sleep(large_timeout) 是否被 Ctrl+C 中断。虽然它仍在轮询，但可能会更早地对 Ctrl+C 做出反应。如果您需要对事件的响应能力，则应将应该对 finished_event.set() 做出反应的代码放入另一个线程中。 - jfs

嗯，即使我不需要响应性，我也倾向于避免轮询，但我想这是唯一的解决方案。 - Navin

警告：while True: pass 会让你的一个 CPU 核心占用率达到 100%。如果你确实需要使用这个解决方案，最好在 pass 前加上 time.sleep(0.1)，以保持最小化的资源消耗。 - Nico Villanueva

@NicoVillanueva：while True: pass 只是一个占位符，用于你真正的代码，它可以执行一些有意义的工作。 - jfs

Answer 3

您还可以通过以下方式修补Event.wait（）函数：

def InterruptableEvent():
    e = threading.Event()

    def patched_wait():
        while not e.is_set():
            e._wait(3)

    e._wait = e.wait
    e.wait = patched_wait
    return e


>>> event = InterruptableEvent()
>>> try:
...     event.wait()
... except KeyboardInterrupt:
...     print "Received KeyboardInterrupt"
... 
^CReceived KeyboardInterrupt

这是因为带有超时参数的wait()会引发KeyboardInterrupt。

- Pieter van den Ham

---

啊，但这仍然像被接受的答案一样在轮询。 - Navin

Answer 4

基于 @Pete 的答案，但是采用子类化并使用实际的 Event.wait 方法，只是将超时时间缩短以允许在处理 KeyboardInterrupt 等情况时进行处理：

class InterruptableEvent(threading.Event):
    def wait(self, timeout=None):
        wait = super().wait  # get once, use often
        if timeout is None:            
            while not wait(0.01):  pass
        else:
            wait(timeout)

- Jeronimo