場景： 如果某個線程持續阻塞，無法退出，從而導致整個程序無法結束，此時就需要強制結束線程

思路：由于程序阻塞，比如卡在代碼中的某一行后，一直無法向下執行，此時，無法通過常規方式結束線程

方法1： 采用線程自己的方法強制結束??Thread._Thread__stop(thd_obj)，此種方式的優點是可以強制結束在某一行代碼處卡住或者正在運行的線程

注：此處的卡住指的是?線程正在發生io阻塞，此時，無法判斷事件，只能在底層加超時處理機制或者采用此處的方式暴力結束，別無他法。

方法2： 使用ctypes的pythonapi接口結束線程，此種方式只能強制結束正在運行的線程而不報錯，無法結束被卡主不動的線程，此種方式無法結束python3的線程

劃重點：python私有變量表示在內的內部定義并使用，外部無法訪問，以雙下劃線作為前作，定義后被python轉為

_classname__變量名

代碼示例1：

importsysimporttimeimportinspectimportctypesimportplatformfrom datetime importdatetimefrom threading importThreaddef_async_raise(tid, exctype):"""raises the exception, performs cleanup if needed"""tid=ctypes.c_long(tid)if notinspect.isclass(exctype):

exctype=type(exctype)

res=ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))if res ==0:raise ValueError("invalid thread id")elif res != 1:#"""if it returns a number greater than one, you're in trouble,

#and you should call it again with exc=NULL to revert the effect"""

ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)raise SystemError("PyThreadState_SetAsyncExc failed")defstop_thread(thd_obj):if platform.python_version() < '3':try:

Thread._Thread.__stop(thd_obj)

#Thread._Thread__stop(thd_obj) # 此種方法也可以

except:pass else: _async_raise(thd_obj.ident, SystemExit)classcommThread(Thread):def __init__(self, target=None, name=None, args=()): super(commThread, self).__init__() self.setName(str(name)) self.fn=target self.args=args self.result= '' defrun(self):print("begin run the child thread： %s" %self.name)whileTrue: time.sleep(1)try:ifself.fn: self.result= self.fn(*self.args)finally:pass defget_result(self):returnself.resultdefget_now_time():return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))def time_str_to_time(st_date, end_date, mod='%Y-%m-%d %H:%M:%S', to_hour=False, to_minute=False, to_second=True): st_date=datetime.strptime(st_date, mod) end_date=datetime.strptime(end_date, mod) time_interval= end_date -st_dateifto_second:returntime_interval.total_seconds()ifto_minute:return ifto_hour:return classA():def __init__(self):pass deftest_ret(self):return 'hello world' deftask1_fn(self, url):whileTrue: time.sleep(1000000)print (url + "FINISHED" +self.test_ret())deftimeout_check(self, st_time):whileTrue: diff=time_str_to_time(st_time, get_now_time())print('check time out is running, diff: %s.' %diff)if diff >= 5:breaktime.sleep(1)returnTrueif __name__ == '__main__': a=A() t1= commThread(target=a.task1_fn, args=('task1_fn',), name='task1_fn') st_time=get_now_time() t2= commThread(target=a.timeout_check, args=(st_time,), name='task2_fn') t1.start() t2.start()whileTrue: t2_result=t2.get_result()print('t2 result: %s' %t2_result)ift2_result:try: stop_thread(t1) stop_thread(t2)#Thread._Thread__stop(t1) #Thread._Thread__stop(t2) #except SystemError as sys_err: #print(sys_err) #except ValueError as val_err: #print(val_err) #except Exception as e: #print(e) except:pass breaktime.sleep(1)print('over .......') sys.exit()

代碼示例2：

fromthreading import Thread

import timeclasscommThread(Thread):

def __init__(self, target=None, name=None, args=()):

super(commThread, self).__init__()

self.setName(str(name))

self.fn=target

self.args=args

self.result= ''def run(self):

print("begin run the child thread: %s" %self.name)whileTrue:

time.sleep(1)try:ifself.fn:

self.result= self.fn(*self.args)finally:

pass

def get_result(self):returnself.result

def kill(self):try:

self._Thread__stop()

except:

pass

def do_scan():whileTrue:

print('scan is running.....')

time.sleep(5)if __name__ == '__main__':for i in range(1):

t1= commThread(target=do_scan, name='do_scan')

t1.setDaemon(True)

t1.start()

t1.join(timeout=3)

print(t1.get_result())ift1.is_alive():

t1.kill()

print('over................')

示例2中充分運用了 setDaemon和join的作用：

1)setDaemon為True表示守護線程或者后臺線程，主線程或者進程退出時，無需等待此線程完成，意義在于：避免子線程無限死循環導致無法退出程序，避免了孤兒進程的出現，設置為False剛好相反，會等待子線程執行完成；

2)join的作用是優先占用cpu資源。join中的timeout表示優先占用cpu資源的最大時長，如果沒有設置，則一直等待直到子線程結束，如果設置了，則達到超時時間后，子線程自動退出。

關于強制結束線程的理解：

python中未提供強制結束線程的接口，原因是這樣做并不安全，

理想的停止退出線程方法是 讓線程自個自殺，所謂的線程自殺就是 我們給線程一個標志位，線程檢測并滿足標志位條件后自己退出。

python的線程不是模擬的，是真實的內核線程，內核調用pthread方法，但Python上層沒有提供關閉線程的方法，這就需要我們自己把握了。

python結束線程的兩種方法：

1) 退出標記

2) 使用ctypes強行殺掉線程

python3強制結束IO阻塞的線程暫未找到合適的方法，如果哪位馬油有方法，還原留言區討論。

總結

以上是生活随笔為你收集整理的python如何强制结束主线程_强制结束线程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。