利用Threading庫我們可以很方便地創(chuàng)建線程，讓它按照我們的想法執(zhí)行我們想讓它執(zhí)行的事情，從而加快程序運行的效率。然而有一點坑爹的是，線程創(chuàng)建之后，就交給了操作系統(tǒng)執(zhí)行，我們無法直接結束一個線程，也無法給它發(fā)送信號，無法調(diào)整它的調(diào)度，也沒有其他高級操作。如果想要相關的功能，只能自己開發(fā)。

怎么開發(fā)呢？

我們創(chuàng)建線程的時候指定了target等于一個我們想讓它執(zhí)行的函數(shù)，這個函數(shù)并不一定是全局函數(shù)，實際上也可以是一個對象中的函數(shù)。如果是對象中的函數(shù)，那么我們就可以在這個函數(shù)當中獲取到對象中的其他信息，我們可以利用這一點來實現(xiàn)手動控制線程的停止。

說起來好像不太好理解，但是看下代碼真的非常簡單：

import timefrom threading import Threadclass TaskWithSwitch: def __init__(self):self._running = True def terminate(self):self._running = False def run(self, n):while self._running and n > 0: print(’Running {}’.format(n)) n -= 1 time.sleep(1)c = TaskWithSwitch()t = Thread(target=c.run, args=(10, ))t.start()c.terminate()t.join()

如果你運行這段代碼，會發(fā)現(xiàn)屏幕上只輸出了10，因為我們將_running這個字段置為False之后，下次循環(huán)的時候不再滿足循環(huán)條件，它就會自己退出了。

如果我們想要用多線程來讀取IO，由于IO可能存在堵塞，所以可能會出現(xiàn)線程一直無法返回的情況。也就是說我們在循環(huán)內(nèi)部卡死了，這個時候單純用_running來判斷還是不夠的，我們需要在線程內(nèi)部設置計時器，防止循環(huán)內(nèi)部的卡死。

class IOTask: def __init__(self):self._running = True def terminate(self):self._running = False def run(self, sock):# 在socket中設置計時器sock.settimeout(10)while self._running: try:# 由于設置了計時器，所以這里不會永久等待data = sock.recv(1024)break except socket.timeout:continuereturn二、線程信號的傳遞

我們之所以如此費勁才能控制線程的運行，主要原因是線程的狀態(tài)是不可知的，并且我們無法直接操作它，因為它是被操作系統(tǒng)管理的。我們運行的主線程和創(chuàng)建出來的線程是獨立的，兩者之間并沒有從屬關系，所以想要實現(xiàn)對線程的狀態(tài)進行控制，往往需要我們通過其他手段來實現(xiàn)。

我們來思考一個場景，假設我們有一個任務，需要在另外一個線程運行結束之后才能開始執(zhí)行。要想要實現(xiàn)這一點，就必須對線程的狀態(tài)有所感知，需要其他線程傳遞出信號來才行。我們可以使用threading中的Event工具來實現(xiàn)這一點。Event工具就是可以用來傳遞信號的，就好像是一個開關，當一個線程執(zhí)行完成之后，會去啟動這個開關。而這個開關控制著另外一段邏輯的運行。

我們來看下樣例代碼：

import timefrom threading import Thread, Eventdef run_in_thread(): time.sleep(1) print(’Thread is running’)t = Thread(target=run_in_thread)t.start()print(’Main thread print’)

我們在線程里面就只做了輸出一行提示符，沒有其他任何邏輯。由于我們在run_in_thread函數(shù)當中沉睡了1s，所以一定是先輸出Main thread print再輸出的Thread is running。假設這個線程是一個很重要的任務，我們希望主線程能夠等待它運行到一個階段再往下執(zhí)行，我們應該怎么辦呢？

注意，這里說的是運行到一個階段，并不是運行結束。運行結束我們很好處理，可以通過join來完成。但如果不是運行結束，而是運行完成了某一個階段，當然通過join也可以，但是會損害整體的效率。這個時候我們就必須要用上Event了。加上Event之后，我們再來看下代碼：

import timefrom threading import Thread, Eventdef run_in_thread(event): time.sleep(1) print(’Thread is running’) # set一下event，這樣外面wait的部分就會被啟動 event.set()# 初始化Eventevent = Event()t = Thread(target=run_in_thread, args=(event, ))t.start()# event等待setevent.wait()print(’Main thread print’)

整體的邏輯沒有太多的修改，主要的是增加了幾行關于Event的使用代碼。

我們?nèi)绻玫紼vent，最好在代碼當中只使用一次。當然通過Event中的clear方法我們可以重置Event的值，但問題是我們沒辦法保證重置的這個邏輯會在wait之前執(zhí)行。如果是在之后執(zhí)行的，那么就會問題，并且在debug的時候會異常痛苦，因為bug不是必現(xiàn)的，而是有時候會出現(xiàn)有時候不會出現(xiàn)。這種情況往往都是因為多線程的使用問題。

所以如果要多次使用開關和信號的話，不要使用Event，可以使用信號量。

Event的問題在于如果多個線程在等待Event的發(fā)生，當它一旦被set的時候，那么這些線程都會同時執(zhí)行。但有時候我們并不希望這樣，我們希望可以控制這些線程一個一個地運行。如果想要做到這一點，Event就無法滿足了，而需要使用信號量。

信號量和Event的使用方法類似，不同的是，信號量可以保證每次只會啟動一個線程。因為這兩者的底層邏輯不太一致，對于Event來說，它更像是一個開關。一旦開關啟動，所有和這個開關關聯(lián)的邏輯都會同時執(zhí)行。而信號量則像是許可證，只有拿到許可證的線程才能執(zhí)行工作，并且許可證一次只發(fā)一張。

想要使用信號量并不需要自己開發(fā)，thread庫當中為我們提供了現(xiàn)成的工具——Semaphore，我們來看它的使用代碼：

# 工作線程def worker(n, sema): # 等待信號量 sema.acquire() print(’Working’, n)# 初始化sema = threading.Semaphore(0)nworkers = 10for n in range(nworkers): t = threading.Thread(target=worker, args=(n, sema,)) t.start()

在上面的代碼當中我們創(chuàng)建了10個線程，雖然這些線程都被啟動了，但是都不會執(zhí)行邏輯，因為sema.acquire是一個阻塞方法，沒有監(jiān)聽到信號量是會一直掛起等待。

當我們釋放信號量之后，線程被啟動，才開始了執(zhí)行。我們每釋放一個信號，則會多啟動一個線程。這里面的邏輯應該不難理解。

在并發(fā)場景當中，多線程的使用絕不是多啟動幾個線程做不同的任務而已，我們需要線程間協(xié)作，需要同步、獲取它們的狀態(tài)，這是非常不容易的。一不小心就會出現(xiàn)幽靈bug，時顯時隱，這也是并發(fā)問題讓人頭疼的主要原因。

以上就是分析Python感知線程狀態(tài)的解決方案之Event與信號量的詳細內(nèi)容，更多關于Python 感知線程狀態(tài) Event與信號量的資料請關注好吧啦網(wǎng)其它相關文章！