802
歡迎關注微信公眾號：羽林君，或者添加作者個人微信：become_me
需求
在python多線程等的使用中，我們需要在外部強制終止線程，這個時候又沒有unix的pthread kill的函數，多進程這個時候大家覺得可以使用kill -9 直接強制殺掉就可以了，從邏輯上這么做沒問題，但是不太優(yōu)雅。其中我總結了一下不僅是使用多線程，以及多協(xié)程、多進程在python的實現對比。
此外也可以參考stackoverlow的文章，如何優(yōu)雅的關閉一個線程，里面有很多的討論，大家可以閱讀一下
下面是網址：https://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread
開始進入正題：
多線程
首先線程中進行退出的話，我們經常會使用一種方式：子線程執(zhí)行的循環(huán)條件設置一個條件，當我們需要退出子線程的時候，將該條件置位，這個時候子線程會主動退出，但是當子線程處于阻塞情況下，沒有在循環(huán)中判斷條件，并且阻塞時間不定的情況下，我們回收該線程也變得遙遙無期。這個時候就需要下面的幾種方式出馬了：
守護線程:
如果你設置一個線程為守護線程，就表示你在說這個線程是不重要的，在進程退出的時候，不用等待這個線程退出。如果你的主線程在退出的時候，不用等待那些子線程完成，那就設置這些線程的daemon屬性。即，在線程開始（thread.start()）之前，調用setDeamon（）函數，設定線程的daemon標志。（thread.setDaemon(True)）就表示這個線程“不重要”。
如果你想等待子線程完成再退出，那就什么都不用做。，或者顯示地調用thread.setDaemon(False)，設置daemon的值為false。新的子線程會繼承父線程的daemon標志。整個Python會在所有的非守護線程退出后才會結束，即進程中沒有非守護線程存在的時候才結束。
也就是子線程為非deamon線程，主線程不立刻退出
import?threading
import?time
import?gc
import?datetime
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.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')?
是否是主線程進行控制？
守護線程需要主線程退出才能完成子線程退出，下面是代碼，再封裝一層進行驗證是否需要主線程退出
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函數執(zhí)行，發(fā)現打印了 stoped 3這個標志后circle線程還在繼續(xù)執(zhí)行。
結論：處理信號靠的就是主線程，只有保證他活著，信號才能正確處理。
在 Python 線程中引發(fā)異常
雖然使用PyThreadState_SetAsyncExc大部分情況下可以滿足我們直接退出線程的操作；但是PyThreadState_SetAsyncExc方法只是為線程退出執(zhí)行“計劃”。它不會殺死線程，尤其是當它正在執(zhí)行外部 C 庫時。嘗試sleep(100)用你的方法殺死一個。它將在 100 秒后被“殺死”。while flag:它與->flag = False方法一樣有效。
所以子線程有例如sleep等阻塞函數時候，在休眠過程中，子線程無法響應，會被主線程捕獲，導致無法取消子線程。就是實際上當線程休眠時候，直接使用async_raise 這個函數殺掉線程并不可以，因為如果線程在 Python 解釋器之外忙，它就不會捕獲中斷
示例代碼：
import?ctypes
import?inspect
import?threading
import?time
import?gc
import?datetime
def?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，可能是后續(xù)版本又做了去除。
這是網上看到的一些示例代碼，但是沒法執(zhí)行，如果有人知道使用可以進行交流。
from?signal?import?pthread_kill,?SIGTSTP
from?threading?import?Thread
from?itertools?import?count
from?time?import?sleep
def?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 包同時提供了本地和遠程并發(fā)操作，通過使用子進程而非線程有效地繞過了 全局解釋器鎖。因此，multiprocessing 模塊允許程序員充分利用給定機器上的多個處理器。
其中使用了multiprocess這些庫，我們可以調用它內部的函數terminate幫我們釋放。例如t.terminate()，這樣就可以強制讓子進程退出了。
不過使用了多進程數據的交互方式比較繁瑣，得使用共享內存、pipe或者消息隊列這些進行子進程和父進程的數據交互。
示例代碼如下：
import?time
import?gc
import?datetime
import?multiprocessing
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?=?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')?
多協(xié)程
協(xié)程(coroutine)也叫微線程，是實現多任務的另一種方式，是比線程更小的執(zhí)行單元，一般運行在單進程和單線程上。因為它自帶CPU的上下文，它可以通過簡單的事件循環(huán)切換任務，比進程和線程的切換效率更高，這是因為進程和線程的切換由操作系統(tǒng)進行。
Python實現協(xié)程的主要借助于兩個庫：asyncio（asyncio 是從Python3.4引入的標準庫，直接內置了對協(xié)程異步IO的支持。asyncio 的編程模型本質是一個消息循環(huán)，我們一般先定義一個協(xié)程函數(或任務), 從 asyncio 模塊中獲取事件循環(huán)loop，然后把需要執(zhí)行的協(xié)程任務(或任務列表)扔到 loop中執(zhí)行，就實現了異步IO）和gevent（Gevent 是一個第三方庫，可以輕松通過gevent實現并發(fā)同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協(xié)程。）。
由于asyncio已經成為python的標準庫了無需pip安裝即可使用，這意味著asyncio作為Python原生的協(xié)程實現方式會更加流行。本文僅會介紹asyncio模塊的退出使用。
使用協(xié)程取消，有兩個重要部分：第一，替換舊的休眠函數為多協(xié)程的休眠函數；第二取消使用cancel()函數。
其中cancel() 返回值為 True 表示 cancel 成功。
示例代碼如下：創(chuàng)建一個coroutine，然后調用run_until_complete()來初始化并啟動服務器來調用main函數，判斷協(xié)程是否執(zhí)行完成，因為設置的num協(xié)程是一個死循環(huán)，所以一直沒有執(zhí)行完，如果沒有執(zhí)行完直接使用 cancel()取消掉該協(xié)程，最后執(zhí)行成功。
import?asyncio
import?time
async?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}被取消")
????????raise
async?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()
結語
這就是我自己的一些python 強制關閉線程、協(xié)程、進程的使用分享。如果大家有更好的想法和需求，也歡迎大家加我好友交流分享哈。
參考文章：在思考如何強制關掉一個子線程或者子進程亦或者協(xié)程時候看了好多文章，python相關的很多信息，方便大家更細節(jié)的查看，我把鏈接放置在下方：
作者：良知猶存，白天努力工作，晚上原創(chuàng)公號號主。公眾號內容除了技術還有些人生感悟，一個認真輸出內容的職場老司機，也是一個技術之外豐富生活的人，攝影、音樂 and 籃球。關注我，與我一起同行。
