Python 殺死執行緒

Vaibhhav Khetarpal 2023年1月30日 2022年5月17日
  1. 在 Python 中線上程中引發異常以終止執行緒
  2. 在 Python 中使用 trace 殺死一個執行緒
  3. 在 Python 中建立/重置停止標誌以終止執行緒
  4. 在 Python 中使用 multiprocessing 模組殺死一個執行緒
  5. 在 Python 中將給定執行緒設定為守護執行緒以終止執行緒
  6. 使用隱藏的 _stop() 函式殺死 Python 中的執行緒
Python 殺死執行緒

儘管它在程式設計師中被標記為一種不好的程式設計習慣,但有時在 Python 中可能仍然需要殺死一個執行緒。本教程演示了我們可以在 Python 中終止執行緒的不同方法。

突然結束威脅的缺點可能會使任務在後臺開啟,從而導致問題。

而且,Python 沒有提供任何直接殺死 Python 中的執行緒的方法,這意味著要找到漏洞和間接方法來實現這一基本任務。

現在,我們將重點介紹並解釋在 Python 中殺死執行緒的幾種方法。

在 Python 中線上程中引發異常以終止執行緒

此方法利用 PyThreadState_SetAsyncExc() 函式,該函式在給定執行緒中非同步引發異常。

以下程式碼線上程中引發異常以終止 Python 中的執行緒。

import threading
import ctypes
import time
class twe(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name        
    def run(self):
        try:
            while True:
                print('running ' + self.name)
        finally:
            print('ended')          
    def get_id(self):
        if hasattr(self, '_thread_id'):
            return self._thread_id
        for id, thread in threading._active.items():
            if thread is self:
                return id
    def raise_exception(self):
        thread_id = self.get_id()
        resu = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id,
              ctypes.py_object(SystemExit))
        if resu > 1: 
            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
            print('Failure in raising exception')      
x = twe('Thread A')
x.start()
time.sleep(2)
x.raise_exception()
x.join()

當程式碼執行時,一旦引發異常,run() 函式將被終止,因為程式控制可以繞過異常處理程式的 try 塊。

然後呼叫 join() 函式以給予最後一擊並殺死 run() 函式。

在 Python 中使用 trace 殺死一個執行緒

在 Python 中實現殺死執行緒的相同任務的另一種方法是在給定執行緒中安裝 trace,改變執行緒的執行。

以下程式碼使用跟蹤來終止 Python 中的執行緒。

import sys
import time
import threading
import trace
class KThread(threading.Thread):
  def __init__(self, *args, **keywords):
    threading.Thread.__init__(self, *args, **keywords)
    self.killed = False
  def start(self):
    self.__run_backup = self.run
    self.run = self.__run     
    threading.Thread.start(self)
  def __run(self):
    sys.settrace(self.globaltrace)
    self.__run_backup()
    self.run = self.__run_backup
  def globaltrace(self, frame, why, arg):
    if why == 'call':
      return self.localtrace
    else:
      return None
  def localtrace(self, frame, why, arg):
    if self.killed:
      if why == 'line':
        raise SystemExit()
    return self.localtrace
  def kill(self):
    self.killed = True
def exfu():
  print('The function begins')
  for i in range(1,100):
    print(i)
    time.sleep(0.2)
  print('The function ends')
x = KThread(target=exfu)
x.start()
time.sleep(1)
x.kill()

上面的程式碼提供了以下輸出。

The function begins
1
2
3
4
5

在這裡,我們使用了 KThread 類,它是原始 threading.Thread 類的一個子集。KThread 類使 kill() 函式在程式碼中實現。

在 Python 中建立/重置停止標誌以終止執行緒

可以在程式碼中宣告一個停止標誌,這將使它在遇到執行緒時停止執行緒的執行。

以下程式碼建立一個停止標誌來終止 Python 中的執行緒。

import threading
import time
def frun():
    while True:
        print('thread running')
        global stop_threads
        if stop_threads:
            break
stop_threads = False
x = threading.Thread(target = frun)
x.start()
time.sleep(1)
stop_threads = True
x.join()
print('killed the thread.')

在 Python 中使用 multiprocessing 模組殺死一個執行緒

multiprocessing 模組使得產生程序成為可能,其方法和工作類似於 threading 模組,因為它們都使用 API。

terminate() 可以殺死一個給定的程序,這比殺死一個執行緒本身更安全,也更簡單。

以下程式碼使用 multiprocessing 模組殺死 Python 中的執行緒。

import multiprocessing
import time
def cp():
  while True:
   for i in range (20):  
    print ('Process: ', i)
    time.sleep(0.05)
x = multiprocessing.Process(target = cp)
x.start()
time.sleep(0.5)
x.terminate()
print("Terminated the child process")

在 Python 中將給定執行緒設定為守護執行緒以終止執行緒

守護執行緒是在主程式終止時自動終止的執行緒。我們可以將給定的執行緒設定為守護執行緒來殺死 Python 中的特定執行緒。

以下程式碼將給定執行緒設定為守護執行緒以終止 Python 中的執行緒。

import threading
import time
import sys
def exfu():
    while True:
        time.sleep(0.5)
        print('Thread alive, but it will die on program termination')
x = threading.Thread(target=exfu)
x.daemon = True
x.start()
time.sleep(2)
sys.exit()

使用隱藏的 _stop() 函式殺死 Python 中的執行緒

儘管沒有記錄,隱藏的 _stop() 函式可以實現在 Python 中殺死執行緒的任務。

以下程式碼使用隱藏的 _stop() 函式來終止 Python 中的執行緒。

import time
import threading
 
class th1(threading.Thread):
    def __init__(self, *args, **kwargs):
        super(th1, self).__init__(*args, **kwargs)
        self._stop = threading.Event()
    def stop(self):
        self._stop.set()
    def stopped(self):
        return self._stop.isSet()
    def run(self):
        while True:
            if self.stopped():
                return
            print("Hello, world!")
            time.sleep(1)
 
x = th1()
x.start()
time.sleep(5)
x.stop()
x.join()
Vaibhhav Khetarpal avatar Vaibhhav Khetarpal avatar

Vaibhhav is an IT professional who has a strong-hold in Python programming and various projects under his belt. He has an eagerness to discover new things and is a quick learner.

LinkedIn

相關文章 - Python Thread