前言:

    不要試圖用強(qiáng)制方法殺掉一個(gè)python線程，這從服務(wù)設(shè)計(jì)上就存在不合理性。 多線程本用來任務(wù)的協(xié)作并發(fā)，如果你使用強(qiáng)制手段干掉線程，那么很大幾率出現(xiàn)意想不到的bug。  請記住一點(diǎn)，鎖資源不會(huì)因?yàn)榫€程退出而釋放鎖資源 ！

我們可以舉出兩個(gè)常見的例子：

1. 有個(gè)A線程拿到了鎖，因?yàn)樗潜粡?qiáng)制干掉的，沒能及時(shí)的release()釋放鎖資源，那么導(dǎo)致所有的線程獲取資源是都被阻塞下去，這就是典型的死鎖場景。

2.在常見的生產(chǎn)消費(fèi)者的場景下，消費(fèi)者從任務(wù)隊(duì)列獲取任務(wù)，但是被干掉后沒有把正在做的任務(wù)丟回隊(duì)列中，那么這就造成了數(shù)據(jù)丟失。

下面是java和python終止線程的方法:

java有三種方法可以使終止線程：

1. 使用退出標(biāo)志，使線程正常退出，也就是當(dāng)run方法完成后線程終止。
2. 使用stop方法強(qiáng)行終止線程（不推薦使用，因?yàn)閟top和suspend、resume一樣，也可能發(fā)生不可預(yù)料的結(jié)果）。
3. 使用interrupt方法中斷線程。

python可以有兩種方法：

1. 退出標(biāo)記
2. 使用ctypes強(qiáng)行殺掉線程

不管是python還是java環(huán)境下，理想的停止退出線程方法是 讓線程自個(gè)自殺，所謂的線程自殺就是 你給他一個(gè)標(biāo)志位，他退出線程。

下面我們會(huì)采用多種方法來測試 停止python線程的異常情況。我們查看一個(gè)進(jìn)程所有的執(zhí)行線程,  進(jìn)程是用過掌控資源，線程是用作調(diào)度單元，進(jìn)程要被調(diào)度執(zhí)行必須要有一個(gè)線程，默認(rèn)的線程和進(jìn)程的pid一樣的。

ps -mp 31449 -o THREAD,tid
 
USER   %CPU PRI SCNT WCHAN USER SYSTEM  TID
root   0.0  -  - -     -   -   -
root   0.0 19  - poll_s  -   - 31449
root   0.0 19  - poll_s  -   - 31450

獲取到了進(jìn)程所有的線程后，通過strace得知 31450 是需要我們kill的線程id，當(dāng)我們kill的時(shí)候，會(huì)出現(xiàn)整個(gè)進(jìn)程都崩潰的情況。 在多線程環(huán)境下，產(chǎn)生的信號是傳遞給整個(gè)進(jìn)程的，一般而言，所有線程都有機(jī)會(huì)收到這個(gè)信號，進(jìn)程在收到信號的的線程上下文執(zhí)行信號處理函數(shù)，具體是哪個(gè)線程執(zhí)行的難以獲知。也就是說，信號會(huì)隨機(jī)發(fā)個(gè)該進(jìn)程的一個(gè)線程。

strace -p <span style="font-size:14px;line-height:21px;">31450</span> Process <span style="font-size:14px;line-height:21px;">31450</span> attached - interrupt to quit
select(0, NULL, NULL, NULL, {0, 320326}) = 0 (Timeout)
select(0, NULL, NULL, NULL, {1, 0})   = 0 (Timeout)
select(0, NULL, NULL, NULL, {1, 0})   = 0 (Timeout)
select(0, NULL, NULL, NULL, {1, 0})   = ? ERESTARTNOHAND (To be restarted)
--- SIGTERM (Terminated) @ 0 (0) ---
Process <span style="font-size:14px;line-height:21px;">31450</span> detached

上面出現(xiàn)的問題其實(shí)跟pthread的說明是一致的。當(dāng)我們在python代碼里加入 signal 信號處理函數(shù)后，回調(diào)函數(shù)可以防止整個(gè)進(jìn)程的退出，那么問題來了，通過信號函數(shù)不能識別你要干掉哪一個(gè)線程，也就是說，不能精準(zhǔn)的干掉某個(gè)線程。你雖然把信號發(fā)給31450線程id，但是信號受理人是所屬進(jìn)程的任何一個(gè)，另外傳給信號處理函數(shù)的參數(shù)只有信號數(shù)和信號stack而已，可有可無的。

加了信號處理后，不會(huì)退出進(jìn)程

select(0, NULL, NULL, NULL, {1, 0})   = 0 (Timeout)
select(0, NULL, NULL, NULL, {1, 0})   = ? ERESTARTNOHAND (To be restarted)
--- SIGTERM (Terminated) @ 0 (0) ---
rt_sigreturn(0xffffffff)        = -1 EINTR (Interrupted system call)
select(0, NULL, NULL, NULL, {1, 0})   = 0 (Timeout)
select(0, NULL, NULL, NULL, {1, 0})   = 0 (Timeout)

如果想從外部通知?dú)⒌裟硞€(gè)線程，那么可以構(gòu)建使用rpc服務(wù)，或者別的方式通信，signal信號不可以，因?yàn)闊o法無法傳遞更多的信息。

python的線程不是模擬的，是真實(shí)的內(nèi)核線程，內(nèi)核調(diào)用pthread方法，但Python上層沒有提供關(guān)閉線程的方法，這就需要我們自己把握了。強(qiáng)烈推薦使用 event 或者 自定義標(biāo)志位的方法， 如果非要強(qiáng)制殺掉線程，那么可以用python ctypes PyThreadState SetAsyncExc 方法強(qiáng)制退出，這樣對于運(yùn)行的python服務(wù)沒有什么影響。

該函數(shù)的實(shí)現(xiàn)原理比較簡單，其實(shí)也是在python虛擬機(jī)里做個(gè)標(biāo)示位，然后由虛擬機(jī)運(yùn)行一個(gè)異常來取消線程，虛擬機(jī)會(huì)幫你做好try cache。 切記不要在外部殺掉python的某個(gè)線程，雖然你能通過ctypes找到線程id，但是你直接kill會(huì)干掉整個(gè)進(jìn)程的。

下面的代碼是 用ctypes 殺掉線程的樣例，不推薦使用，因?yàn)樘直┝?

import ctypes
 
def terminate_thread(thread):
  if not thread.isAlive():
    return
 
  exc = ctypes.py_object(SystemExit)
  res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
    ctypes.c_long(thread.ident), exc)
  if res == 0:
    raise ValueError("nonexistent thread id")
  elif res > 1:
    ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
    raise SystemError("PyThreadState_SetAsyncExc failed")

咱們簡單look一下PyThreadState源代碼，總而言之觸發(fā)線程的異常模式。 有興趣的人可以閱讀 python pystate.c 的設(shè)計(jì)，配合著youtube的一些視頻分享。

 
int
PyThreadState_SetAsyncExc(long id, PyObject *exc) {
  PyInterpreterState *interp = GET_INTERP_STATE();
  ...
  HEAD_LOCK();
  for (p = interp->tstate_head; p != NULL; p = p->next) {
    if (p->thread_id == id) {
      從鏈表里找到線程的id，避免死鎖，我們需要釋放head_mutex。
      PyObject *old_exc = p->async_exc;
      Py_XINCREF(exc); #增加該對象的引用數(shù)
      p->async_exc = exc; # 更為exc模式
      HEAD_UNLOCK();
      Py_XDECREF(old_exc); # 因?yàn)橐∠?dāng)然也就遞減引用
      ...
      return 1; #銷毀線程成功
    }
  }
  HEAD_UNLOCK();
  return 0;
}

原生posix pthread 可以使用 ptread_cancel(tid) 在主線程中結(jié)束子線程。但是 Python 的線程庫不支持這樣做，理由是我們不應(yīng)該強(qiáng)制地結(jié)束一個(gè)線程，這樣會(huì)帶來很多隱患，應(yīng)該讓該線程自己結(jié)束自己。所以在 Python 中，推薦的方法是在子線程中循環(huán)判斷一個(gè)標(biāo)志位，在主線程中改變該標(biāo)志位，子線程讀到標(biāo)志位改變，就結(jié)束自己。

類似這個(gè)邏輯:

def consumer_threading():
 t1_stop= threading.Event()
 t1 = threading.Thread(target=thread1, args=(1, t1_stop))
 
 t2_stop = threading.Event()
 t2 = threading.Thread(target=thread2, args=(2, t2_stop))
 
 time.sleep(duration)
 #stop the thread2
 t2_stop.set()
 
def thread1(arg1, stop_event):
 while(not stop_event.is_set()):
   #similar to time.sleep()
   stop_event.wait(time)
   pass
 
 
def thread2(arg1, stop_event):
 while(not stop_event.is_set()):
   stop_event.wait(time)
   pass

簡單的總結(jié)，雖然我們可以用ctypes里的pystats來控制線程，但這種粗暴中斷線程的方法是不合理的。 請選用 自殺模式 ！如果你的線程正在發(fā)生io阻塞，而不能判斷事件怎么辦？ 你的程序需要做優(yōu)化了，最少在網(wǎng)絡(luò)io層需要有主動(dòng)的timeout，避免一直的阻塞下去。

最新評論