場(chǎng)景： 如果某個(gè)線(xiàn)程持續(xù)阻塞，無(wú)法退出，從而導(dǎo)致整個(gè)程序無(wú)法結(jié)束，此時(shí)就需要強(qiáng)制結(jié)束線(xiàn)程

思路：由于程序阻塞，比如卡在代碼中的某一行后，一直無(wú)法向下執(zhí)行，此時(shí)，無(wú)法通過(guò)常規(guī)方式結(jié)束線(xiàn)程

方法1： 采用線(xiàn)程自己的方法強(qiáng)制結(jié)束??Thread._Thread__stop(thd_obj)，此種方式的優(yōu)點(diǎn)是可以強(qiáng)制結(jié)束在某一行代碼處卡住或者正在運(yùn)行的線(xiàn)程

注：此處的卡住指的是?線(xiàn)程正在發(fā)生io阻塞，此時(shí)，無(wú)法判斷事件，只能在底層加超時(shí)處理機(jī)制或者采用此處的方式暴力結(jié)束，別無(wú)他法。

方法2： 使用ctypes的pythonapi接口結(jié)束線(xiàn)程，此種方式只能強(qiáng)制結(jié)束正在運(yùn)行的線(xiàn)程而不報(bào)錯(cuò)，無(wú)法結(jié)束被卡主不動(dòng)的線(xiàn)程，此種方式無(wú)法結(jié)束python3的線(xiàn)程

劃重點(diǎn)：python私有變量表示在內(nèi)的內(nèi)部定義并使用，外部無(wú)法訪(fǎng)問(wèn)，以雙下劃線(xiàn)作為前作，定義后被python轉(zhuǎn)為

_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中充分運(yùn)用了 setDaemon和join的作用：

1)setDaemon為T(mén)rue表示守護(hù)線(xiàn)程或者后臺(tái)線(xiàn)程，主線(xiàn)程或者進(jìn)程退出時(shí)，無(wú)需等待此線(xiàn)程完成，意義在于：避免子線(xiàn)程無(wú)限死循環(huán)導(dǎo)致無(wú)法退出程序，避免了孤兒進(jìn)程的出現(xiàn)，設(shè)置為False剛好相反，會(huì)等待子線(xiàn)程執(zhí)行完成；

2)join的作用是優(yōu)先占用cpu資源。join中的timeout表示優(yōu)先占用cpu資源的最大時(shí)長(zhǎng)，如果沒(méi)有設(shè)置，則一直等待直到子線(xiàn)程結(jié)束，如果設(shè)置了，則達(dá)到超時(shí)時(shí)間后，子線(xiàn)程自動(dòng)退出。

關(guān)于強(qiáng)制結(jié)束線(xiàn)程的理解：

python中未提供強(qiáng)制結(jié)束線(xiàn)程的接口，原因是這樣做并不安全，

理想的停止退出線(xiàn)程方法是 讓線(xiàn)程自個(gè)自殺，所謂的線(xiàn)程自殺就是 我們給線(xiàn)程一個(gè)標(biāo)志位，線(xiàn)程檢測(cè)并滿(mǎn)足標(biāo)志位條件后自己退出。

python的線(xiàn)程不是模擬的，是真實(shí)的內(nèi)核線(xiàn)程，內(nèi)核調(diào)用pthread方法，但Python上層沒(méi)有提供關(guān)閉線(xiàn)程的方法，這就需要我們自己把握了。

python結(jié)束線(xiàn)程的兩種方法：

1) 退出標(biāo)記

2) 使用ctypes強(qiáng)行殺掉線(xiàn)程

python3強(qiáng)制結(jié)束IO阻塞的線(xiàn)程暫未找到合適的方法，如果哪位馬油有方法，還原留言區(qū)討論。

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。