使用 ctrl + c 停止 Python 脚本

Stop Python script with ctrl + c

我有一个使用线程的脚本,但它无法捕获 Ctrl + C。

这是重现此错误的示例代码:

import threading
import time
import signal

class DummyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

    def stop(self, signum=None, frame=None):
        self._running = False

    def run(self):
        while self._running:
            time.sleep(1)
            print("Running")

if __name__ == "__main__":
    try:
        t = DummyThread()
        t.start()
        while True:
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("This never gets printed")
        t.stop()
    finally:
        print("Exit")

当我 运行 python3 script.py 它开始 运行ning,但它没有捕捉到 ctrl+c。我用谷歌搜索了它,但我还没有找到解决方案。我必须用 SIGTERM 终止脚本,但我希望 DummyThread 正常停止。

要点是您不能在除主线程之外的任何其他线程中使用信号。主线程是唯一可以接收信号并处理它们的线程。我可以提供以下解决方案,它基于 Event 同步原语。

According to Python documantation:

Signals and threads Python signal handlers are always executed in the main Python thread, even if the signal was received in another thread. This means that signals can’t be used as a means of inter-thread communication. You can use the synchronization primitives from the threading module instead.

此外,只允许主线程设置新的信号处理程序。

from threading import Thread, Event
import time


class DummyThread(Thread):

    def __init__(self, event: Event):
        Thread.__init__(self)
        self.stop_event = event

    def run(self):
        # we are monitoring the event in the Main Thread
        while not self.stop_event.is_set():  
            time.sleep(1)
            print("Running")
        # only Main Thread can make the point reachable
        print("I am done !")  


if __name__ == "__main__":
    try:
        e = Event() 
        t = DummyThread(e)
        t.start()
        while True:
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:
        e.set()
    finally:
        print("Exit")

另一种可能的选择是使用 daemon 线程来完成代码示例中的此类任务(当您每秒在屏幕上打印 smth,而不是例如关闭数据库连接或某些类似任务时)。如果主线程停止,daemon 线程也会停止。

如您的代码所示,您使用 KeyboardInterrupt 调用了 stop() 函数。查看 Listener 如何执行相同的任务并停止无法从 Ctrl + C 捕获的执行。您不必再使用 SIGTERM

终止脚本
import threading
import time
import signal
import os
from pynput.keyboard import Key, Listener

class DummyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

    def stop(self, signum=None, frame=None):
        self._running = False
        print ("Bye Bye . .")
        os._exit(1)

    def run(self):
        while self._running:
            time.sleep(1)
            print("Running")


if __name__ == "__main__":
    t = DummyThread()
    def func2():
        try:
            t.start()
            while True:
                print("Main thread running")
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("No need for this")
            t.stop()
        finally:
            print("Exit")

    def func1():
        with Listener(on_press = t.stop) as listener :
            listener.join()

    threading.Thread(target=func1).start()
    threading.Thread(target=func2).start()
class DummyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)
        self._running = True
        signal.signal(signal.SIGINT, self.stop)
        signal.signal(signal.SIGTERM, self.stop)

该程序实际上没有按预期工作因为最后两行并且没有它们也可以工作。

原因是,如果您按下 Ctrl-CSIGINT 信号由 signal.signal 设置的信号处理程序处理self.stop 被调用。所以线程实际上应该停止。

但是在main线程中,while True循环仍然是运行。由于信号已被处理,Python 运行时将 没有 KeyboardInterrupt 引发异常 。因此,您永远不会到达 except 部分。

if __name__ == "__main__":
    try:
        t = DummyThread()
        t.start()
        while True:            # you are stuck in this loop
            print("Main thread running")
            time.sleep(0.5)
    except KeyboardInterrupt:  # this never happens
        print("This never gets printed")
        t.stop()

只应设置一个信号处理程序来调用 stop 方法。所以解决问题的方案有两种:

  1. 通过捕获 KeyboardInterrupt 异常隐式处理信号。这是通过简单地删除两个 signal.signal(...) 行来实现的。

  2. 设置一个明确的信号处理程序(就像你在DummyThread.__init__中使用signal.signal所做的那样),但是从主线程中删除while True:循环并执行不要尝试处理 KeyboardInterrupt。相反,只需等待 DummyThread 使用其 join 方法自行完成:

    if __name__ == "__main__":
        t = DummyThread()
        t.start()
        t.join()
        print("Exit")