一、简单说明

将启动、终止和查看线程状态的方法封装成类
声明时传入要启动的方法
通过 start、stop 和 state 执行启动、终止和查看状态

二、实现步骤

# encoding: utf-8import time
import threading
import inspect
import ctypesclass MyThreadFunc(object):'''手动终止线程的方法'''def __init__(self, func, argsTup):self.myThread = threading.Thread(target=func, args=argsTup)def start(self):print('线程启动')self.myThread.start()def state(self):status = self.myThread.is_alive()print('线程状态: {0}'.format(status))return statusdef stop(self):print('线程终止')try:for i in range(5):self._async_raise(self.myThread.ident, SystemExit)time.sleep(1)except Exception as e:print(e)def _async_raise(self, tid, exctype):"""raises the exception, performs cleanup if needed"""tid = ctypes.c_long(tid)if not inspect.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")if __name__ == '__main__':# 定义一个读秒器def second_count(arg1, arg2):i = 0while True:i += 1print("{}-{}: {}".format(arg1, arg2, i))time.sleep(1)# 声明一个线程类mythread = MyThreadFunc(second_count, ("好耶", "haoye"))# 启动 --------------------------------------mythread.start()# 等待三秒time.sleep(3)# 查看线程状态mythread.state()# 等待三秒time.sleep(3)# 终止线程 ----------------------------------mythread.stop()# 等待三秒time.sleep(3)# 再次查看线程状态mythread.state()

三、测试

在这里插入图片描述
ps：python3.2 之后可以使用 concurrent.futures，该模块在 python 的多线程 threading、多进程multiprocesssing 上进一步封装，实现了进程池和线程池，可以参考这篇文章 https://blog.csdn.net/weixin_43721000/article/details/128177331