PHP前端开发

python中强制关闭线程、协程与进程的方法是什么

百变鹏仔 1个月前 (01-22) #Python
文章标签 线程

多线程

首先线程中进行退出的话,我们经常会使用一种方式:子线程执行的循环条件设置一个条件,当我们需要退出子线程的时候,将该条件置位,这个时候子线程会主动退出,但是当子线程处于阻塞情况下,没有在循环中判断条件,并且阻塞时间不定的情况下,我们回收该线程也变得遥遥无期。这个时候就需要下面的几种方式出马了:

守护线程:

如果你设置一个线程为守护线程,就表示你在说这个线程是不重要的,在进程退出的时候,不用等待这个线程退出。
如果你的主线程在退出的时候,不用等待那些子线程完成,那就设置这些线程的daemon属性。即,在线程开始(thread.start())之前,调用setDeamon()函数,设定线程的daemon标志。(thread.setDaemon(True))就表示这个线程“不重要”。

如果你想等待子线程完成再退出,那就什么都不用做。,或者显示地调用thread.setDaemon(False),设置daemon的值为false。新的子线程会继承父线程的daemon标志。整个Python会在所有的非守护线程退出后才会结束,即进程中没有非守护线程存在的时候才结束。

也就是子线程为非deamon线程,主线程不立刻退出

import threadingimport timeimport gcimport datetimedef circle():    print("begin")    try:        while True:            current_time = datetime.datetime.now()            print(str(current_time) + ' circle.................')            time.sleep(3)    except Exception as e:        print('error:',e)    finally:        print('end')if __name__ == "__main__":   t = threading.Thread(target=circle)   t.setDaemon(True)   t.start()   time.sleep(1)   # stop_thread(t)   # print('stoped threading Thread')    current_time = datetime.datetime.now()   print(str(current_time) + ' stoped after')    gc.collect()   while True:      time.sleep(1)      current_time = datetime.datetime.now()      print(str(current_time) + ' end circle')

是否是主线程进行控制?

立即学习“Python免费学习笔记(深入)”;

守护线程需要主线程退出才能完成子线程退出,下面是代码,再封装一层进行验证是否需要主线程退出

def Daemon_thread():   circle_thread= threading.Thread(target=circle)#    circle_thread.daemon = True   circle_thread.setDaemon(True)   circle_thread.start()      while running:        print('running:',running)         time.sleep(1)   print('end..........') if __name__ == "__main__":    t = threading.Thread(target=Daemon_thread)    t.start()       time.sleep(3)    running = False    print('stop running:',running)     print('stoped 3')     gc.collect()    while True:        time.sleep(3)        print('stoped circle')

替换main函数执行,发现打印了 stoped 3这个标志后circle线程还在继续执行。

结论:处理信号靠的就是主线程,只有保证他活着,信号才能正确处理。

在 Python 线程中引发异常

虽然使用PyThreadState_SetAsyncExc大部分情况下可以满足我们直接退出线程的操作;但是PyThreadState_SetAsyncExc方法只是为线程退出执行“计划”。它不会杀死线程,尤其是当它正在执行外部 C 库时。尝试sleep(100)用你的方法杀死一个。它将在 100 秒后被“杀死”。while flag:它与->flag = False方法一样有效。

所以子线程有例如sleep等阻塞函数时候,在休眠过程中,子线程无法响应,会被主线程捕获,导致无法取消子线程。就是实际上当线程休眠时候,直接使用async_raise 这个函数杀掉线程并不可以,因为如果线程在 Python 解释器之外忙,它就不会捕获中断

示例代码:

import ctypesimport inspectimport threadingimport timeimport gcimport datetimedef async_raise(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")      def stop_thread(thread):   async_raise(thread.ident, SystemExit)   def circle():    print("begin")    try:        while True:            current_time = datetime.datetime.now()            print(str(current_time) + ' circle.................')            time.sleep(3)    except Exception as e:        print('error:',e)    finally:        print('end')if __name__ == "__main__":   t = threading.Thread(target=circle)   t.start()   time.sleep(1)   stop_thread(t)   print('stoped threading Thread')    current_time = datetime.datetime.now()   print(str(current_time) + ' stoped after')    gc.collect()   while True:      time.sleep(1)      current_time = datetime.datetime.now()      print(str(current_time) + ' end circle')

signal.pthread_kill操作:

这个是最接近与 unix中pthread kill操作,网上看到一些使用,但是自己验证时候没有找到这个库里面的使用,

这是在python官方的signal解释文档里面的描述,看到是3.3 新版功能,我自己本身是python3.10,没有pthread_kill,可能是后续版本又做了去除。

这是网上看到的一些示例代码,但是没法执行,如果有人知道使用可以进行交流。

from signal import pthread_kill, SIGTSTPfrom threading import Threadfrom itertools import countfrom time import sleepdef target():    for num in count():        print(num)        sleep(1)thread = Thread(target=target)thread.start()sleep(5)signal.pthread_kill(thread.ident, SIGTSTP)

多进程

multiprocessing 是一个支持使用与 threading 模块类似的 API 来产生进程的包。 multiprocessing 包同时提供了本地和远程并发操作,通过使用子进程而非线程有效地绕过了 全局解释器锁。 因此,multiprocessing 模块允许程序员充分利用给定机器上的多个处理器。

其中使用了multiprocess这些库,我们可以调用它内部的函数terminate帮我们释放。例如t.terminate(),这样就可以强制让子进程退出了。

不过使用了多进程数据的交互方式比较繁琐,得使用共享内存、pipe或者消息队列这些进行子进程和父进程的数据交互。

示例代码如下:

import timeimport gcimport datetimeimport multiprocessingdef circle():    print("begin")    try:        while True:            current_time = datetime.datetime.now()            print(str(current_time) + ' circle.................')            time.sleep(3)    except Exception as e:        print('error:',e)    finally:        print('end')if __name__ == "__main__":    t = multiprocessing.Process(target=circle, args=())    t.start()    # Terminate the process    current_time = datetime.datetime.now()    print(str(current_time) + ' stoped before')     time.sleep(1)    t.terminate()  # sends a SIGTERM    current_time = datetime.datetime.now()    print(str(current_time) + ' stoped after')     gc.collect()    while True:        time.sleep(3)        current_time = datetime.datetime.now()        print(str(current_time) + ' end circle')

多协程

协程(coroutine)也叫微线程,是实现多任务的另一种方式,是比线程更小的执行单元,一般运行在单进程和单线程上。因为它自带CPU的上下文,它可以通过简单的事件循环切换任务,比进程和线程的切换效率更高,这是因为进程和线程的切换由操作系统进行。

Python实现协程的主要借助于两个库:asyncio(asyncio 是从Python3.4引入的标准库,直接内置了对协程异步IO的支持。asyncio 的编程模型本质是一个消息循环,我们一般先定义一个协程函数(或任务), 从 asyncio 模块中获取事件循环loop,然后把需要执行的协程任务(或任务列表)扔到 loop中执行,就实现了异步IO)和gevent(Gevent 是一个第三方库,可以轻松通过gevent实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。)。

由于asyncio已经成为python的标准库了无需pip安装即可使用,这意味着asyncio作为Python原生的协程实现方式会更加流行。本文仅会介绍asyncio模块的退出使用。

使用协程取消,有两个重要部分:第一,替换旧的休眠函数为多协程的休眠函数;第二取消使用cancel()函数。

其中cancel() 返回值为 True 表示 cancel 成功。

示例代码如下:创建一个coroutine,然后调用run_until_complete()来初始化并启动服务器来调用main函数,判断协程是否执行完成,因为设置的num协程是一个死循环,所以一直没有执行完,如果没有执行完直接使用 cancel()取消掉该协程,最后执行成功。

import asyncioimport timeasync def num(n):    try:        i = 0        while True:            print(f'i={i} Hello')            i=i+1            # time.sleep(10)            await asyncio.sleep(n*0.1)        return n    except asyncio.CancelledError:        print(f"数字{n}被取消")        raiseasync def main():    # tasks = [num(i) for i in range(10)]    tasks = [num(10)]    complete, pending = await asyncio.wait(tasks, timeout=0.5)    for i in complete:        print("当前数字",i.result())    if pending:        print("取消未完成的任务")        for p in pending:            p.cancel()if __name__ == '__main__':    loop = asyncio.get_event_loop()    try:        loop.run_until_complete(main())    finally:        loop.close()