python如何强制结束主线程

python如何强制结束主线程_强制结束线程

场景：如果某个线程持续阻塞，无法退出，从而导致整个程序无法结束，此时就需要强制结束线程

思路：由于程序阻塞，比如卡在代码中的某一行后，一直无法向下执行，此时，无法通过常规方式结束线程

方法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: