搞事情 -- python之线程
簡介
- 操作系統線程理論
- 線程概念的引入背景
- 線程的特點
- 進程和線程的關系
- 使用線程的實際場景
- 用戶級線程和內核級線程(了解)
- 線程和python
- 理論知識
- 線程的創建Threading.Thread類
- 鎖
- 信號量
- 事件
- 條件
- 定時器
- 隊列
- Python標準模塊--concurrent.futures
操作系統線程理論
線程概念的引入背景
進程
之前我們已經了解了操作系統中進程的概念，程序并不能單獨運行，只有將程序裝載到內存中，系統為它分配資源才能運行，而這種執行的程序就稱之為進程。程序和進程的區別就在于：程序是指令的集合，它是進程運行的靜態描述文本；進程是程序的一次執行活動，屬于動態概念。在多道編程中，我們允許多個程序同時加載到內存中，在操作系統的調度下，可以實現并發地執行。正是這樣的設計，大大提高了CPU的利用率。進程的出現讓每個用戶感覺到自己獨享CPU，因此，進程就是為了在CPU上實現多道編程而提出的。
進程這么牛逼，為什么還要有線程？
進程有很多優點，它提供了多道編程，讓我們感覺我們每個人都擁有自己的CPU和其他資源，可以提高計算機的利用率。很多人就不理解了，既然進程這么優秀，為什么還要線程呢？其實，仔細觀察就會發現進程還是有很多缺陷的，主要體現在兩點上：
? ? ? ?1 ?進程只能在一個時間干一件事，如果想同時干兩件事或多件事，進程就無能為力了。
? ? ? ?2 ?進程在執行的過程中如果阻塞，例如等待輸入，整個進程就會掛起，即使進程中有些工作不依賴于輸入的數據，也將無法執行。
如果這兩個缺點理解比較困難的話，舉個現實的例子也許你就清楚了：如果把我們上課的過程看成一個進程的話，那么我們要做的是耳朵聽老師講課，手上還要記筆記，腦子還要思考問題，這樣才能高效的完成聽課的任務。而如果只提供進程這個機制的話，上面這三件事將不能同時執行，同一時間只能做一件事，聽的時候就不能記筆記，也不能用腦子思考，這是其一；如果老師在黑板上寫演算過程，我們開始記筆記，而老師突然有一步推不下去了，阻塞住了，他在那邊思考著，而我們呢，也不能干其他事，即使你想趁此時思考一下剛才沒聽懂的一個問題都不行，這是其二。
現在你應該明白了進程的缺陷了，而解決的辦法很簡單，我們完全可以讓聽、寫、思三個獨立的過程，并行起來，這樣很明顯可以提高聽課的效率。而實際的操作系統中，也同樣引入了這種類似的機制——線程。
線程的出現
60年代，在OS中能擁有資源和獨立運行的基本單位是進程，然而隨著計算機技術的發展，進程出現了很多弊端，一是由于進程是資源擁有者，創建、撤消與切換存在較大的時空開銷，因此需要引入輕型進程；二是由于對稱多處理機（SMP）出現，可以滿足多個運行單位，而多個進程并行開銷過大。 因此在80年代，出現了能獨立運行的基本單位——線程（Threads）。 注意：進程是資源分配的最小單位,線程是CPU調度的最小單位. 每一個進程中至少有一個線程。進程和線程的關系
線程與進程的區別可以歸納為以下4點： 1）地址空間和其它資源（如打開文件）：進程間相互獨立，同一進程的各線程間共享。某進程內的線程在其它進程不可見。 2）通信：進程間通信IPC，線程間可以直接讀寫進程數據段（如全局變量）來進行通信——需要進程同步和互斥手段的輔助，以保證數據的一致性。 3）調度和切換：線程上下文切換比進程上下文切換要快得多。 4）在多線程操作系統中，進程不是一個可執行的實體。 *通過漫畫了解線程進城
線程的特點
在多線程的操作系統中，通常是在一個進程中包括多個線程，每個線程都是作為利用CPU的基本單位，是花費最小開銷的實體。線程具有以下屬性。 1）輕型實體 線程中的實體基本上不擁有系統資源，只是有一點必不可少的、能保證獨立運行的資源。 線程的實體包括程序、數據和TCB。線程是動態概念，它的動態特性由線程控制塊TCB（Thread Control Block）描述。TCB包括以下信息： （1）線程狀態。 （2）當線程不運行時，被保存的現場資源。 （3）一組執行堆棧。 （4）存放每個線程的局部變量主存區。 （5）訪問同一個進程中的主存和其它資源。 用于指示被執行指令序列的程序計數器、保留局部變量、少數狀態參數和返回地址等的一組寄存器和堆棧。TCB包含以下信息 2）獨立調度和分派的基本單位。 在多線程OS中，線程是能獨立運行的基本單位，因而也是獨立調度和分派的基本單位。由于線程很“輕”，故線程的切換非常迅速且開銷小（在同一進程中的）。 3）共享進程資源。 線程在同一進程中的各個線程，都可以共享該進程所擁有的資源，這首先表現在：所有線程都具有相同的進程id，這意味著，線程可以訪問該進程的每一個內存資源；此外，還可以訪問進程所擁有的已打開文件、定時器、信號量機構等。由于同一個進程內的線程共享內存和文件，所以線程之間互相通信不必調用內核。 4）可并發執行。 在一個進程中的多個線程之間，可以并發執行，甚至允許在一個進程中所有線程都能并發執行；同樣，不同進程中的線程也能并發執行，充分利用和發揮了處理機與外圍設備并行工作的能力。
使用線程的實際場景
?
開啟一個字處理軟件進程，該進程肯定需要辦不止一件事情，比如監聽鍵盤輸入，處理文字，定時自動將文字保存到硬盤，這三個任務操作的都是同一塊數據，因而不能用多進程。只能在一個進程里并發地開啟三個線程,如果是單線程，那就只能是，鍵盤輸入時，不能處理文字和自動保存，自動保存時又不能輸入和處理文字。
內存中的線程
?
多個線程共享同一個進程的地址空間中的資源，是對一臺計算機上多個進程的模擬，有時也稱線程為輕量級的進程。
而對一臺計算機上多個進程，則共享物理內存、磁盤、打印機等其他物理資源。多線程的運行也和多進程的運行類似，是cpu在多個線程之間的快速切換。
不同的進程之間是充滿敵意的，彼此是搶占、競爭cpu的關系，如果迅雷會和QQ搶資源。而同一個進程是由一個程序員的程序創建，所以同一進程內的線程是合作關系，一個線程可以訪問另外一個線程的內存地址，大家都是共享的，一個線程干死了另外一個線程的內存，那純屬程序員腦子有問題。
類似于進程，每個線程也有自己的堆棧，不同于進程，線程庫無法利用時鐘中斷強制線程讓出CPU，可以調用thread_yield運行線程自動放棄cpu，讓另外一個線程運行。
線程通常是有益的，但是帶來了不小程序設計難度，線程的問題是：
1. 父進程有多個線程，那么開啟的子線程是否需要同樣多的線程
2. 在同一個進程中，如果一個線程關閉了文件，而另外一個線程正準備往該文件內寫內容呢？
因此，在多線程的代碼中，需要更多的心思來設計程序的邏輯、保護程序的數據。
用戶級線程和內核級線程（了解）
線程的實現可以分為兩類：用戶級線程(User-Level Thread)和內核線線程(Kernel-Level Thread)，后者又稱為內核支持的線程或輕量級進程。在多線程操作系統中，各個系統的實現方式并不相同，在有的系統中實現了用戶級線程，有的系統中實現了內核級線程。?
用戶級線程
內核的切換由用戶態程序自己控制內核切換,不需要內核干涉，少了進出內核態的消耗，但不能很好的利用多核Cpu。
在用戶空間模擬操作系統對進程的調度，來調用一個進程中的線程，每個進程中都會有一個運行時系統，用來調度線程。此時當該進程獲取cpu時，進程內再調度出一個線程去執行，同一時刻只有一個線程執行。
內核級線程
?內核級線程:切換由內核控制，當線程進行切換的時候，由用戶態轉化為內核態。切換完畢要從內核態返回用戶態；可以很好的利用smp，即利用多核cpu。windows線程就是這樣的。
用戶級與內核級線程的對比
1 內核支持線程是OS內核可感知的，而用戶級線程是OS內核不可感知的。 2 用戶級線程的創建、撤消和調度不需要OS內核的支持，是在語言（如Java）這一級處理的；而內核支持線程的創建、撤消和調度都需OS內核提供支持，而且與進程的創建、撤消和調度大體是相同的。 3 用戶級線程執行系統調用指令時將導致其所屬進程被中斷，而內核支持線程執行系統調用指令時，只導致該線程被中斷。 4 在只有用戶級線程的系統內，CPU調度還是以進程為單位，處于運行狀態的進程中的多個線程，由用戶程序控制線程的輪換運行；在有內核支持線程的系統內，CPU調度則以線程為單位，由OS的線程調度程序負責線程的調度。 5 用戶級線程的程序實體是運行在用戶態下的程序，而內核支持線程的程序實體則是可以運行在任何狀態下的程序。用戶級線程和內核級線程的區別用戶級線程和內核級線程的區別
?
優點：當有多個處理機時，一個進程的多個線程可以同時執行。
缺點：由內核進行調度。 內核線程的優缺點 ?
優點：
線程的調度不需要內核直接參與，控制簡單。
可以在不支持線程的操作系統中實現。
創建和銷毀線程、線程切換代價等線程管理的代價比內核線程少得多。
允許每個進程定制自己的調度算法，線程管理比較靈活。
線程能夠利用的表空間和堆棧空間比內核級線程多。
同一進程中只能同時有一個線程在運行，如果有一個線程使用了系統調用而阻塞，那么整個進程都會被掛起。另外，頁面失效也會產生同樣的問題。
缺點：
資源調度按照進程進行，多個處理機下，同一個進程中的線程只能在同一個處理機下分時復用 用戶級線程的優缺點 ?
混合實現
用戶級與內核級的多路復用，內核同一調度內核線程，每個內核線程對應n個用戶線程
linux操作系統的 NPTL
歷史 在內核2.6以前的調度實體都是進程，內核并沒有真正支持線程。它是能過一個系統調用clone()來實現的，這個調用創建了一份調用進程的拷貝，跟fork()不同的是,這份進程拷貝完全共享了調用進程的地址空間。LinuxThread就是通過這個系統調用來提供線程在內核級的支持的(許多以前的線程實現都完全是在用戶態，內核根本不知道線程的存在)。非常不幸的是，這種方法有相當多的地方沒有遵循POSIX標準，特別是在信號處理，調度，進程間通信原語等方面。很顯然，為了改進LinuxThread必須得到內核的支持，并且需要重寫線程庫。為了實現這個需求，開始有兩個相互競爭的項目：IBM啟動的NGTP(Next Generation POSIX Threads)項目，以及Redhat公司的NPTL。在2003年的年中，IBM放棄了NGTP，也就是大約那時，Redhat發布了最初的NPTL。NPTL最開始在redhat linux 9里發布，現在從RHEL3起內核2.6起都支持NPTL，并且完全成了GNU C庫的一部分。設計 NPTL使用了跟LinuxThread相同的辦法，在內核里面線程仍然被當作是一個進程，并且仍然使用了clone()系統調用(在NPTL庫里調用)。但是，NPTL需要內核級的特殊支持來實現，比如需要掛起然后再喚醒線程的線程同步原語futex.NPTL也是一個1*1的線程庫，就是說，當你使用pthread_create()調用創建一個線程后，在內核里就相應創建了一個調度實體，在linux里就是一個新進程，這個方法最大可能的簡化了線程的實現。除NPTL的1*1模型外還有一個m*n模型，通常這種模型的用戶線程數會比內核的調度實體多。在這種實現里，線程庫本身必須去處理可能存在的調度，這樣在線程庫內部的上下文切換通常都會相當的快，因為它避免了系統調用轉到內核態。然而這種模型增加了線程實現的復雜性,并可能出現諸如優先級反轉的問題，此外，用戶態的調度如何跟內核態的調度進行協調也是很難讓人滿意。介紹介紹
?
python和線程
理論知識
全局解釋器鎖GIL
Python代碼的執行由Python虛擬機(也叫解釋器主循環)來控制。Python在設計之初就考慮到要在主循環中，同時只有一個線程在執行。雖然 Python 解釋器中可以“運行”多個線程，但在任意時刻只有一個線程在解釋器中運行。
對Python虛擬機的訪問由全局解釋器鎖(GIL)來控制，正是這個鎖能保證同一時刻只有一個線程在運行。
在多線程環境中，Python 虛擬機按以下方式執行：
a、設置 GIL；
b、切換到一個線程去運行；
c、運行指定數量的字節碼指令或者線程主動讓出控制(可以調用 time.sleep(0))；
d、把線程設置為睡眠狀態；
e、解鎖 GIL；
d、再次重復以上所有步驟。
在調用外部代碼(如 C/C++擴展函數)的時候，GIL將會被鎖定，直到這個函數結束為止(由于在這期間沒有Python的字節碼被運行，所以不會做線程切換)編寫擴展的程序員可以主動解鎖GIL。
python線程模塊的選擇
Python提供了幾個用于多線程編程的模塊，包括thread、threading和Queue等。thread和threading模塊允許程序員創建和管理線程。thread模塊提供了基本的線程和鎖的支持，threading提供了更高級別、功能更強的線程管理的功能。Queue模塊允許用戶創建一個可以用于多個線程之間共享數據的隊列數據結構。
避免使用thread模塊，因為更高級別的threading模塊更為先進，對線程的支持更為完善，而且使用thread模塊里的屬性有可能會與threading出現沖突；其次低級別的thread模塊的同步原語很少(實際上只有一個)，而threading模塊則有很多；再者，thread模塊中當主線程結束時，所有的線程都會被強制結束掉，沒有警告也不會有正常的清除工作，至少threading模塊能確保重要的子線程退出后進程才退出。?
thread模塊不支持守護線程，當主線程退出時，所有的子線程不論它們是否還在工作，都會被強行退出。而threading模塊支持守護線程，守護線程一般是一個等待客戶請求的服務器，如果沒有客戶提出請求它就在那等著，如果設定一個線程為守護線程，就表示這個線程是不重要的，在進程退出的時候，不用等待這個線程退出。
threading模塊
multiprocess模塊的完全模仿了threading模塊的接口，二者在使用層面，有很大的相似性，因而不再詳細介紹（官方鏈接）
需注意，玩多線程編程時，可以不用再像寫多進程一樣，每個py文件中必須加上if __name__ == '__main__'這一行代碼了
線程的創建Threading.Thread類
線程的創建
from threading import Thread import timedef func():print('你瞅啥')if __name__ == '__main__':t = Thread(target=func)t.start()time.sleep(1)print('瞅你咋地')常規版
from threading import Thread import timeclass MyThread(Thread):def __init__(self):super(MyThread, self).__init__()def run(self):time.sleep(1)print('就是這么diao')t = MyThread() t.start()逼格版
多線程與多進程
from threading import Thread from multiprocessing import Process import os import time def son(n):print('%s中，我是兒子，我的pid號是: %s'%(n,os.getpid()))if __name__ == '__main__':t = Thread(target=son,args=('線程',))t.start()print('線程中，我是爸爸，我的pid號是：%s'%os.getpid())time.sleep(2)p = Process(target=son,args=('進程',))p.start()print('進程中，我是爸爸，我的pid號是：%s' % os.getpid())關于pid的比較
?
from threading import Thread from multiprocessing import Process import os import time def son(n):passif __name__ == '__main__':start = time.time()for i in range(100):t = Thread(target=son,args=('線程',))t.start()print(time.time() - start)for i in range(100):p = Process(target=son,args=('進程',))p.start()print(time.time() - start)關于開啟效率的對比
?
from threading import Thread import timedef func():global numnum -= 1time.sleep(2)if __name__ == '__main__':num = 100t = Thread(target=func,)t.start()t.join()print(num)# 很明顯，多線程是共享一個進程內部的全局變量的# 進程就不開啟了，已經學習過進程了，多進程就是不能共享數據關于共享數據的比較
?
多線程的socket聊天from threading import Thread import socketsk = socket.socket() sk.bind(('127.0.0.1',8080)) sk.listen()def func(conn):while 1:info = conn.recv(1024).decode('utf-8')print(info)conn.send(info.upper().encode('utf-8'))if __name__ == '__main__':while 1:conn,addr = sk.accept()Thread(target=func,args=(conn,)).start()TCP協議服務器
import socketsk = socket.socket() sk.connect(('127.0.0.1',8080))while 1:msg_s = input('>>>')sk.send(msg_s.encode('utf-8'))print(sk.recv(1024).decode('utf-8'))TCP協議客戶端
?
Thread類的其他方法
Thread實例對象的方法# isAlive(): 返回線程是否活動的。# getName(): 返回線程名。# setName(): 設置線程名。 threading模塊提供的一些方法：# threading.currentThread(): 返回當前的線程對象。# threading.current_thread():返回當前的線程對象，和上述方法一樣。# threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啟動后、結束前，不包括啟動前和終止后的線程。# threading.activeCount(): 返回正在運行的線程數量，與len(threading.enumerate())有相同的結果。# threading.active_count():返回正在運行的線程數量，和上述方法一樣。更多方法供你用
from threading import Thread import threading import timedef func():time.sleep(2)if __name__ == '__main__':for i in range(20):t = Thread(target=func)t.start()# print(t.getName())# t.setName('alex')# print(t.getName())# print(t.isAlive())print(threading.currentThread())print(threading.current_thread())print(threading.enumerate())print(threading.activeCount())print(threading.active_count())方法示例
from threading import Thread import threading import timedef func():time.sleep(2)if __name__ == '__main__':for i in range(20):t = Thread(target=func)t.start()t.join()#讓主線程停留在這一行，等待子線程結束join示例
?
守護線程
無論是進程還是線程，都遵循：守護xx會等待主xx運行完畢后被銷毀。需要強調的是：運行完畢并非終止運行
#1.對主進程來說，運行完畢指的是主進程代碼運行完畢 #2.對主線程來說，運行完畢指的是主線程所在的進程內所有非守護線程統統運行完畢，主線程才算運行完畢
#1 主進程在其代碼結束后就已經算運行完畢了（守護進程在此時就被回收）,然后主進程會一直等非守護的子進程都運行完畢后回收子進程的資源(否則會產生僵尸進程)，才會結束， #2 主線程在其他非守護線程運行完畢后才算運行完畢（守護線程在此時就被回收）。因為主線程的結束意味著進程的結束，進程整體的資源都將被回收，而進程必須保證非守護線程都運行完畢后才能結束。詳解
from threading import Thread import timedef func():time.sleep(2)if __name__ == '__main__':t = Thread(target=func)t.setDaemon(True)t.start()print('主線程，守護線程是否還存活？',t.is_alive())守護線程示例1
from threading import Thread import timedef func():time.sleep(2)print('這里是子線程，就是守護線程，就是這么diao，還活著呢！')def func1():time.sleep(4)print('這里是子線程，并不是守護線程')if __name__ == '__main__':t = Thread(target=func)t1 = Thread(target=func1)t.setDaemon(True)t.start()t1.start()print('這里是父線程')守護線程示例2
鎖
鎖與GIL
同步鎖
from threading import Thread import timedef func():global ntmp = ntime.sleep(0.1)n = tmp - 1t_l = [] n = 100 for i in range(100):t = Thread(target=func)t.start()t_l.append(t)[t.join() for t in t_l] print(n)# 結果極有可能是99顛覆你三觀的一個代碼
import threading R=threading.Lock() R.acquire() ''' 對公共數據的操作 ''' R.release()
from threading import Thread,Lock import time # 注意，同步鎖又叫做互斥鎖def func():global nl.acquire()tmp = ntime.sleep(0.1)n = tmp - 1l.release()t_l = [] n = 100 l = Lock() for i in range(100):t = Thread(target=func)t.start()t_l.append(t)[t.join() for t in t_l] print(n)# 結果肯定是0，之前的代碼沒加鎖，100個線程是異步執行。此時加了鎖，100個線程想要操作變量n，必須同步的去操作同步鎖 -- 互斥鎖
大家有沒有發現剛才代碼，既然我用了加鎖的方式，讓子線程變成了串行操作。那么肯定有人就有疑問了，那么我在for循環中干脆直接讓子線程join住不就行了，也是讓子線程變成了串行執行。
我告訴你，你的這個思想很有深度，確實，你這種方式確實也可以保證數據的安全性。
但是，記不記得我之前說過，加鎖就盡可能的小范圍加鎖，這樣就會使子線程中有一部分代碼是并行的，有一部分代碼是串行的。而如果使用join的話，就會讓子線程整個代碼都是串行，毫無疑問，雖然使用join也能讓子線程變成串行并且能保護數據安全性，但是相對加鎖來說，效率又變低下了很多。小伙伴感興趣的可以去試試效率。 用帶著疑問的眼光去看問題 ?
死鎖與遞歸鎖
進程也有死鎖與遞歸鎖，在進程那里忘記說了，放到這里一切說了額
所謂死鎖： 是指兩個或兩個以上的進程或線程在執行過程中，因爭奪資源而造成的一種互相等待的現象，若無外力作用，它們都將無法推進下去。此時稱系統處于死鎖狀態或系統產生了死鎖，這些永遠在互相等待的進程稱為死鎖進程，如下就是最簡單的死鎖
from threading import Lock as Lock import time mutexA=Lock() mutexA.acquire() mutexA.acquire() print(123) mutexA.release() mutexA.release()死鎖的簡單實現
?
解決方法，遞歸鎖，在Python中為了支持在同一線程中多次請求同一資源，python提供了可重入鎖RLock。
這個RLock內部維護著一個Lock和一個counter變量，counter記錄了acquire的次數，從而使得資源可以被多次acquire。直到一個線程所有的acquire都被release，其他的線程才能獲得資源。上面的例子如果使用RLock代替Lock，則不會發生死鎖：
from threading import RLock as Lock import time mutexA=Lock() mutexA.acquire() mutexA.acquire() print(123) mutexA.release() mutexA.release()遞歸鎖RLock遞歸鎖RLock
?死鎖問題：兩個人（兩個進程），都想上廁所，但是男的先搶到了廁所資源，男的進入廁所后發現沒帶紙。此時女的在廁所外邊沒獲得廁所資源，但是她有紙資源。互相都想獲取互相的資源，造成死鎖
from threading import Thread,Lock import timedef Man():l_tot.acquire()print('男同志以百米沖刺的速度，搶到了廁所資源！')time.sleep(1)l_wsz.acquire()print('男同志以秋風掃落葉般的氣勢獲得了衛生紙資源。')time.sleep(1)print('男同志可以安心的解決自己的事情了')l_wsz.release()l_tot.release()def Women():l_wsz.acquire()print('女同志以秋風掃落葉般的氣勢獲得了衛生紙資源。')time.sleep(1)l_tot.acquire()print('女同志以百米沖刺的速度，搶到了廁所資源！')time.sleep(1)print('女同志可以安心的解決自己的事情了')l_tot.release()l_wsz.release()if __name__ == '__main__':l_tot = Lock()l_wsz = Lock()man_t = Thread(target=Man)man_t.start()woman_t = Thread(target=Women)woman_t.start()死鎖
from threading import Thread,RLock import timedef Man():l_tot.acquire()print('男同志以百米沖刺的速度，搶到了廁所資源！')time.sleep(1)l_wsz.acquire()print('男同志以秋風掃落葉般的氣勢獲得了衛生紙資源。')time.sleep(1)print('男同志可以安心的解決自己的事情了')l_wsz.release()l_tot.release()def Women():l_wsz.acquire()print('女同志以秋風掃落葉般的氣勢獲得了衛生紙資源。')time.sleep(1)l_tot.acquire()print('女同志以百米沖刺的速度，搶到了廁所資源！')time.sleep(1)print('女同志可以安心的解決自己的事情了')l_tot.release()l_wsz.release()if __name__ == '__main__':l_tot =l_wsz = RLock()man_t = Thread(target=Man)man_t.start()woman_t = Thread(target=Women)woman_t.start()遞歸鎖解決死鎖問題
?
信號量
同進程的一樣
Semaphore管理一個內置的計數器，
每當調用acquire()時內置計數器-1；
調用release() 時內置計數器+1；
計數器不能小于0；當計數器為0時，acquire()將阻塞線程直到其他線程調用release()。
實例：(同時只有5個線程可以獲得semaphore,即可以限制最大連接數為5)：
from threading import Thread,Semaphore import time import randomdef func(i):sem.acquire()print('第%s個客人進入屋子'%i)time.sleep(random.randint(1,3))print('第%s個客人離開屋子' % i)sem.release()sem = Semaphore(5) for i in range(20):t = Thread(target=func,args=(i,))t.start()舉個栗子
?
事件
同進程的一樣
線程的一個關鍵特性是每個線程都是獨立運行且狀態不可預測。如果程序中的其 他線程需要通過判斷某個線程的狀態來確定自己下一步的操作,這時線程同步問題就會變得非常棘手。為了解決這些問題,我們需要使用threading庫中的Event對象。 對象包含一個可由線程設置的信號標志,它允許線程等待某些事件的發生。在 初始情況下,Event對象中的信號標志被設置為假。如果有線程等待一個Event對象, 而這個Event對象的標志為假,那么這個線程將會被一直阻塞直至該標志為真。一個線程如果將一個Event對象的信號標志設置為真,它將喚醒所有等待這個Event對象的線程。如果一個線程等待一個已經被設置為真的Event對象,那么它將忽略這個事件, 繼續執行。
event.isSet()：返回event的狀態值； event.wait()：如果 event.isSet()==False將阻塞線程； event.set()： 設置event的狀態值為True，所有阻塞池的線程激活進入就緒狀態， 等待操作系統調度； event.clear()：恢復event的狀態值為False。
?
例如，有多個工作線程嘗試鏈接MySQL，我們想要在鏈接前確保MySQL服務正常才讓那些工作線程去連接MySQL服務器，如果連接不成功，都會去嘗試重新連接。那么我們就可以采用threading.Event機制來協調各個工作線程的連接操作。from threading import Thread,Event import time import randomdef conn_mysql(e,i):count = 1while 1:if e.is_set():# 如果為True，就是可以連接上數據庫breakif count > 3:print('連接超時')returnprint('第%s個人正在嘗試第%s次連接!'%(i,count))e.wait(0.5)# 在這里阻塞等待0.5秒，模擬用戶連接時的等待count+=1print('第%s個人連接成功'%i)def check_mysql(e):print('\033[45m 數據庫正在維護 \033[0m')# 讓數據庫初始狀態處于維護狀態，默認所有用戶連接不上time.sleep(random.randint(1,2))# 隨機1秒或2秒，如果隨機1秒的話，用戶就可以連接上，2秒就連接不上e.set()# 將e.is_set()設為Trueif __name__ == '__main__':e = Event()t = Thread(target=check_mysql,args=(e,))t.start()for i in range(10):# 產生10個線程都去嘗試連接數據庫t1 = Thread(target=conn_mysql,args=(e,i))t1.start()模擬連接數據庫操作
?
條件
使得線程等待，只有滿足某條件時，才釋放n個線程
Python提供的Condition對象提供了對復雜線程同步問題的支持。Condition被稱為條件變量，除了提供與Lock類似的acquire和release方法外，還提供了wait和notify方法。線程首先acquire一個條件變量，然后判斷一些條件。如果條件不滿足則wait；如果條件滿足，進行一些處理改變條件后，通過notify方法通知其他線程，其他處于wait狀態的線程接到通知后會重新判斷條件。不斷的重復這一過程，從而解決復雜的同步問題。
from threading import Condition,Thread# 條件# 條件被創建之初，默認有一個False狀態(只是舉個例子，其實沒有什么狀態) # False狀態會影響wait，會讓wait一直處于阻塞等待狀態，等待notify制造鑰匙。 # notify(int) 是制造鑰匙的意思，要制造幾把鑰匙def func(con,i):con.acquire()con.wait()# 所有線程都阻塞在這里，因為還沒制造出鑰匙print('在第%s個循環里'%i)con.release()#并不是還鑰匙，而是把鑰匙扔掉if __name__ == '__main__':con = Condition()for i in range(10):t = Thread(target=func,args=(con,i))t.start()while 1:num = int(input('>>>'))con.acquire()con.notify(num)# 手動輸入制造幾把鑰匙，相當于發一個信號，允許幾個線程可以執行了con.release()舉個例子
定時器
定時器，指定n秒后執行某個操作
from threading import Timerdef func():print("開始計時收費了")t = Timer(1, func) t.start() # after 1 seconds, "hello, world" will be printed
?
線程隊列
queue隊列 ：使用import queue，用法與進程Queue一樣
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
- class?
queue.Queue(maxsize=0) #先進先出import queueq=queue.Queue() q.put('first') q.put('second') q.put('third')print(q.get()) print(q.get()) print(q.get())先進先出型
?class?queue.LifoQueue(maxsize=0) #last in fisrt out
import queueq=queue.LifoQueue() q.put('first') q.put('second') q.put('third')print(q.get()) print(q.get()) print(q.get())先進后出型
?class?queue.PriorityQueue(maxsize=0) #存儲數據時可設置優先級的隊列
import queueq=queue.PriorityQueue() #put進入一個元組,元組的第一個元素是優先級(通常是數字,也可以是非數字之間的比較),數字越小優先級越高 q.put((20,'a')) q.put((10,'b')) q.put((30,'c'))print(q.get()) print(q.get()) print(q.get())# 很明顯，數字越小優先級越高優先級隊列
Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).exception queue.Empty Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.exception queue.Full Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.Queue.qsize() Queue.empty() #return True if empty Queue.full() # return True if full Queue.put(item, block=True, timeout=None) Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).Queue.put_nowait(item) Equivalent to put(item, False).Queue.get(block=True, timeout=None) Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).Queue.get_nowait() Equivalent to get(False).Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.Queue.task_done() Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).Raises a ValueError if called more times than there were items placed in the queue.Queue.join() block直到queue被消費完畢更多方法說明更多方法
?
Python標準模塊--concurrent.futures
https://docs.python.org/dev/library/concurrent.futures.html
#1 介紹 concurrent.futures模塊提供了高度封裝的異步調用接口 ThreadPoolExecutor：線程池，提供異步調用 ProcessPoolExecutor: 進程池，提供異步調用 Both implement the same interface, which is defined by the abstract Executor class.#2 基本方法 #submit(fn, *args, **kwargs) 異步提交任務#map(func, *iterables, timeout=None, chunksize=1) 取代for循環submit的操作#shutdown(wait=True) 相當于進程池的pool.close()+pool.join()操作 wait=True，等待池內所有任務執行完畢回收完資源后才繼續 wait=False，立即返回，并不會等待池內的任務執行完畢 但不管wait參數為何值，整個程序都會等到所有任務執行完畢 submit和map必須在shutdown之前#result(timeout=None) 取得結果#add_done_callback(fn) 回調函數
?
#介紹 The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None) An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.from concurrent.futures import ProcessPoolExecutor# 計算平方求和 def func(n):sum = 0for i in range(n):sum += n ** 2return sumif __name__ == '__main__':pp = ProcessPoolExecutor(max_workers=5)pp_l = []for i in range(15):s = pp.submit(func,i)pp_l.append(s)pp.shutdown()[print(s.result()) for s in pp_l]ProcessPoolExecutor舉個栗子
#介紹 ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously. class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='') An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.#用法 與ProcessPoolExecutor相同ThreadPoolExecutorThreadPoolExecutor和上述用法一樣
from concurrent.futures import ThreadPoolExecutordef func(n):sum = 0for i in range(n):sum += i**2return sumif __name__ == '__main__':tt = ThreadPoolExecutor(max_workers=5)# for i in range(15):# tt.submit(func,i)s = tt.map(func,range(15))print(s.__next__())print(s.__next__())print(s.__next__())print(s.__next__())map的用法及返回值
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor from multiprocessing import Pool import requests import os,timedef get_page(url):print('<進程%s> get %s' %(os.getpid(),url))respone=requests.get(url)if respone.status_code == 200:return {'url':url,'text':respone.text}def parse_page(res):res=res.result()print('<進程%s> parse %s' %(os.getpid(),res['url']))parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))with open('db.txt','a') as f:f.write(parse_res)if __name__ == '__main__':urls=['https://www.baidu.com','https://www.python.org','https://www.openstack.org','https://help.github.com/','http://www.sina.com.cn/']# p=Pool(5)# start = time.time()# for url in urls:# p.apply_async(get_page,args=(url,),callback=parse_page)# p.close()# p.join()# print(time.time() - start) # 測試Pool進程池的效率 start = time.time()# p=ThreadPoolExecutor(5)p=ProcessPoolExecutor(5)for url in urls:p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一個future對象obj，需要用obj.result()拿到結果 p.shutdown()print(time.time() - start)# 分別測試ProcessPoolExecutor和ThreadPoolExecutor的效率回調函數
?
轉載于:https://www.cnblogs.com/LY-C/p/9482766.html
總結
以上是生活随笔為你收集整理的搞事情 -- python之线程的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 贵州省有哪几个机场~？
- 下一篇： 杜教