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

【问题标题】：Catch Keyboard Interrupt in program that is waiting on an Event在等待事件的程序中捕获键盘中断
【发布时间】：2013-01-03 11:23:06
【问题描述】：

以下程序会挂起终端以忽略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()

【问题讨论】：

标签： python windows multithreading windows-7 python-3.x

【解决方案1】：

更新：在当前的 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 相关的bugs。测试它是否适用于您的环境。

^{旧的投票答案：}

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

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

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

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

【讨论】：

我不知道。它适用于轮询吗？为什么民意调查不好？您可以测试time.sleep(large_timeout) 是否被Ctrl+C 打断。尽管它仍在轮询，但它可能会更快地对Ctrl+C 做出反应。在这种情况下，如果您需要对事件做出响应，应该将应该对 finished_event.set() 做出反应的代码放入另一个线程中。
好吧，即使我并不真正需要响应，我也倾向于避免轮询，但我想这是唯一的解决方案。
while True: pass 的警告词：这将使您的 CPU 核心之一达到 100%。如果你真的需要使用这个方案，最好在pass之前抛出一个time.sleep(0.1)，以将消耗降到最低。
@NicoVillanueva：while True: pass 只是您的真实代码的占位符，可以做一些有意义的工作。

【解决方案2】：

如果要避免轮询，可以使用signal模块的pause()函数代替finished_event.wait()。 signal.pause() 是一个阻塞函数，当进程接收到信号时会解除阻塞。在这种情况下，当按下 ^C 时，SIGINT 信号会解除对函数的阻塞。请注意，根据文档，该功能在 Windows 上不起作用。我已经在 Linux 上尝试过，它对我有用。

我在这个 SO thread 中遇到了这个解决方案。

【讨论】：

有点晚了，但感谢您的非投票回答:)
我认为你可以编辑你的答案更简单直接，但你解决了我的问题，谢谢！

【解决方案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。

【讨论】：

啊，但这仍然像接受的答案一样轮询:(

【解决方案4】：

基于@Pete 的回答，但使用子类化并使用实际的Event.wait 方法，只是使用较小的超时来处理KeyboardInterrupts 等：

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)

【讨论】：