快捷導(dǎo)航

python中停止線(xiàn)程的方法代碼舉例

更新時(shí)間：2024年05月21日 11:09:18 作者：邏輯峰

在Python中停止線(xiàn)程有多種方法,包括使用全局變量、使用標(biāo)志位、使用異常等,下面這篇文章主要給大家介紹了關(guān)于python中停止線(xiàn)程方法的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下

1 threading.Event()方法

一種常見(jiàn)的方法是使用標(biāo)志位來(lái)通知線(xiàn)程應(yīng)該停止。線(xiàn)程可以定期檢查這個(gè)標(biāo)志位，如果它被設(shè)置為停止，那么線(xiàn)程就結(jié)束其執(zhí)行。下面是一個(gè)簡(jiǎn)單的例子：

import threading  
import time  
  
class MyThread(threading.Thread):  
    def __init__(self):  
        super(MyThread, self).__init__()  
        self.stop_event = threading.Event()  
  
    def run(self):  
    	# 清空設(shè)置
        self.stop_event.clear()
        while not self.stop_event.is_set():  
            print("Thread is running...")  
            time.sleep(1)  
  
    def stop(self):  
        self.stop_event.set()  
  
# 創(chuàng)建線(xiàn)程  
thread = MyThread()  
thread.start()  
  
# 在某個(gè)時(shí)間點(diǎn)，停止線(xiàn)程  
time.sleep(5)  
thread.stop()

需要注意的是，這只是一種優(yōu)雅的停止線(xiàn)程的方法，它依賴(lài)于線(xiàn)程在run方法中定期檢查stop_event。如果線(xiàn)程沒(méi)有這樣的檢查，或者它正在執(zhí)行一個(gè)無(wú)法被中斷的阻塞操作（例如IO操作），那么這種方法可能無(wú)法立即停止線(xiàn)程。

2 子線(xiàn)程拋出異常，立刻停止

在Python中，使用ctypes和PyThreadState_SetAsyncExc函數(shù)來(lái)在子線(xiàn)程中異步拋出一個(gè)異常是一種相對(duì)底層的做法，它可以直接在子線(xiàn)程的上下文中觸發(fā)一個(gè)異常。然而，這種做法需要謹(jǐn)慎使用，因?yàn)樗赡軙?huì)導(dǎo)致線(xiàn)程狀態(tài)不穩(wěn)定或未定義的行為，特別是如果線(xiàn)程沒(méi)有正確地處理異常。

import threading
import time
import ctypes
import inspect

def do_some_task():
    while True:
        time.sleep(1)
        print("子線(xiàn)程！1")
        time.sleep(1)
        print("子線(xiàn)程！2")
        time.sleep(1)
        print("子線(xiàn)程！3")
        time.sleep(1)
        print("子線(xiàn)程！4")
        time.sleep(1)
        print("子線(xiàn)程！5")

def async_raise(thread_id, exctype):
    """
    通過(guò)C語(yǔ)言的庫(kù)拋出異常
    :param thread_id:
    :param exctype:
    :return:
    """
    # 在子線(xiàn)程內(nèi)部拋出一個(gè)異常結(jié)束線(xiàn)程
    thread_id = ctypes.c_long(thread_id)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("線(xiàn)程id違法")
    elif res != 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, None)
        raise SystemError("異常拋出失敗")

def stop_thread_now(thread):
    # 結(jié)束線(xiàn)程
    async_raise(thread.ident, SystemExit)

if __name__ == "__main__":
    # 可以在子線(xiàn)程任何時(shí)候隨時(shí)結(jié)束子線(xiàn)程
    sub_thread = threading.Thread(target=do_some_task,name="sub_thread")
    sub_thread.start()
    print(sub_thread.is_alive())
    time.sleep(7)
    stop_thread_now(sub_thread)
    time.sleep(1)
    print(sub_thread.is_alive())

附：線(xiàn)程停止的最佳實(shí)踐

在實(shí)際的應(yīng)用中，我們應(yīng)該遵循以下最佳實(shí)踐來(lái)停止線(xiàn)程：

使用標(biāo)志位或Event對(duì)象來(lái)控制線(xiàn)程的停止，這樣可以在線(xiàn)程執(zhí)行的關(guān)鍵點(diǎn)進(jìn)行安全的停止。
在線(xiàn)程的代碼中定期檢查標(biāo)志位或Event對(duì)象的狀態(tài)，并根據(jù)狀態(tài)決定是否退出循環(huán)。
在主線(xiàn)程中修改標(biāo)志位或Event對(duì)象的狀態(tài)時(shí)，需要使用線(xiàn)程鎖來(lái)保證線(xiàn)程安全性。
避免使用Thread的stop()方法來(lái)終止線(xiàn)程的執(zhí)行。