如何在Python中優雅地終止線程，什麼時候需要強制終止？

優雅的線程終止

通常不鼓勵突然終止線程，尤其是在 Python 中。如果關鍵操作中斷，可能會導致資源洩漏或資料損壞。

建議方法

首選方法是設定一個標誌或訊號量來指示執行緒應該退出。線程應定期檢查此標誌，如果已設置，則正常退出。

範例：

import threading

class StoppableThread(threading.Thread):
    def __init__(self,  *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

在此範例中，呼叫 stop() 向執行緒發出訊號退出，然後使用 join() 等待它優雅地完成。

強制終止

在特殊情況下，您可能需要強制終止執行緒。但是，這應該被視為最後的手段。

強制終止方法：

import ctypes
import inspect

def _async_raise(tid, exctype):
    if not inspect.isclass(exctype):
        raise TypeError("Only types can be raised (not instances)")
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class ThreadWithExc(threading.Thread):
    def _get_my_tid(self):
        if not self.is_alive(): # Note: self.isAlive() on older version of Python
            raise threading.ThreadError("the thread is not active")

        # do we have it cached?
        if hasattr(self, "_thread_id"):
            return self._thread_id

        # no, look for it in the _active dict
        for tid, tobj in threading._active.items():
            if tobj is self:
                self._thread_id = tid
                return tid

        raise AssertionError("could not determine the thread's id")

    def raise_exc(self, exctype):
        _async_raise(self._get_my_tid(), exctype )

此方法依賴 PyThreadState_SetAsyncExc 函數在特定的環境中引發異常線。但是，需要注意的是，此方法並不完全可靠，如果執行緒處於 Python 解釋器之外的系統呼叫中，則可能會失敗。

注意事項：

• 盡量避免使用此方法。
• 如有必要，實作線程特定的清理邏輯以確保資料誠信。
• 謹慎使用此方法，了解潛在的風險和限制。

以上是如何在Python中優雅地終止線程，什麼時候需要強制終止？的詳細內容。更多資訊請關注PHP中文網其他相關文章！