如何从Python线程退出整个应用程序？

Question

如何从其中一个线程退出整个 Python 应用程序？

sys.exit()

仅终止调用它的线程，因此没有帮助。

我不想使用

os.kill()

解决方案，因为这不是很干净。

Answer 1

简短回答：使用

os._exit

。

长答案并举例：

我从 DevShed 上的教程中提取并稍微修改了一个简单的线程示例：

import threading, sys, os

theVar = 1

class MyThread ( threading.Thread ):

   def run ( self ):

      global theVar
      print 'This is thread ' + str ( theVar ) + ' speaking.'
      print 'Hello and good bye.'
      theVar = theVar + 1
      if theVar == 4:
          #sys.exit(1)
          os._exit(1)
      print '(done)'

for x in xrange ( 7 ):
   MyThread().start()

如果将

sys.exit(1)

注释掉，脚本将在第三个线程打印出来后终止。如果您使用

sys.exit(1)

并注释掉

os._exit(1)

，则第三个线程不会打印 (done)

，并且程序将运行所有七个线程。

os._exit

“通常只应在 fork() 之后的子进程中使用”——并且一个单独的线程足以满足您的目的。另请注意，该手册页中 

os._exit

 之后列出了几个枚举值，您应该更喜欢将它们作为 

os._exit

 的参数，而不是像我在上面的示例中使用的简单数字。

Answer 2

如果除主线程之外的所有线程都是守护进程，则最好的方法通常是

thread.interrupt_main()——任何线程都可以使用它在主线程中引发 KeyboardInterrupt

，这通常可以导致相当干净的退出来自主线程（包括主线程中的终结器被调用等）。

当然，如果这导致一些非守护线程保持整个进程处于活动状态，您需要按照 Mark 的建议使用

os._exit

 进行后续操作 - 但我认为这是最后的手段（有点像 

kill -9

 ;-) 因为它非常粗暴地终止事物（终结器不运行，包括 

try/finally

 块、

with

 块、

atexit

 函数等）。

Answer 3

在某些情况下使用

thread.interrupt_main()

 可能没有帮助。 

KeyboardInterrupt

经常在命令行应用程序中用于退出当前命令或清理输入行。

此外，

os._exit

将立即终止进程，而不运行代码中的任何

finally

块，这可能是危险的（例如文件和连接不会关闭）。

我找到的解决方案是在主线程中注册一个引发自定义异常的信号处理程序。使用后台线程来触发信号。

import signal
import os
import threading
import time


class ExitCommand(Exception):
    pass


def signal_handler(signal, frame):
    raise ExitCommand()


def thread_job():
    time.sleep(5)
    os.kill(os.getpid(), signal.SIGUSR1)


signal.signal(signal.SIGUSR1, signal_handler)
threading.Thread(target=thread_job).start()  # thread will fire in 5 seconds
try:
    while True:
        user_input = raw_input('Blocked by raw_input loop ')
        # do something with 'user_input'
except ExitCommand:
    pass
finally:
    print('finally will still run')

相关问题：

• 为什么在 Python 线程内调用 sys.exit() 时不退出？
• Python：当陷入阻塞 raw_input 时如何退出 CLI？

Answer 4

退出整个程序最简单的方法是，我们应该使用进程ID（pid）来终止程序。

import os
import psutil

current_system_pid = os.getpid()

ThisSystem = psutil.Process(current_system_pid)
ThisSystem.terminate()

安装 psutl：-“pip install psutil”

Answer 5

对于 Linux，您可以使用

kill()

 命令并传递当前进程的 ID 和 SIGINT 信号来启动退出应用程序的步骤。

import signal

os.kill(os.getpid(), signal.SIGINT)

Answer 6

使用threading.Event()

它允许您处理错误并适用于 Windows

和 Linux。

每当达到所需阈值时，使用此脚本结束所有线程。在这种情况下，我将其用作超时。

其他解决方案依赖于可能有问题的 os._exit，这是故意不这样做的，以便您可以使用 try/ except/finally 块来处理异常，而不仅仅是退出。它也适用于 Python 2

和 3，但许多答案并不适用。

此脚本将等到您的任一函数完成后再完成。

import time, queue, threading

def my_func(s):  # Function that you want to set a time limit for.
    # Insert your code here. Sleep is used to mimic operation time.
    print('Running my_func for {}s'.format(s))
    time.sleep(s)
    print('my_func complete {}s'.format(time.time()-t0))
    e.set()

def timeout_func(s):  # Controls timeout
    print('Starting timeout_func for {}s'.format(s))
    time.sleep(s)
    print('timeout_func complete {}s'.format(time.time()-t0))
    e.set()

t0 = time.time()
q = queue.Queue()
myfunc_runtime = 5  # Mimics the amount of time your function runs
timeout_threshold = 10  # Seconds until timeout

# Initialize and start threads
thread_timeout = threading.Thread(target=timeout_func, args=[timeout_threshold])
thread_myfunc = threading.Thread(target=my_func, args=[myfunc_runtime])
thread_timeout.daemon = True  # Essential for event to stop thread
thread_myfunc.daemon = True  # Essential for event to stop thread
e = threading.Event()
thread_timeout.start()
thread_myfunc.start()

# Wait until your event is set. Script completion time is always the minimum of runtime for both funcs
e.wait()
print('Script completion time: ' + str(time.time()-t0))

Answer 7

另一个解决方案（非常简单，但计算效率不高）是将其添加到主线程的无限 while 循环中：

if not other_thread_name.is_alive():
  exit()

所以你基本上是定期检查另一个线程是否已经退出。如果有，主线程也会退出。