python之路——線程

簡介

• 操作系統(tǒng)線程理論
  • 線程概念的引入背景
  • 線程的特點(diǎn)
  • 進(jìn)程和線程的關(guān)系
  • 使用線程的實(shí)際場景
  • 用戶級線程和內(nèi)核級線程(了解)
• 線程和python
  • 理論知識
  • 線程的創(chuàng)建Threading.Thread類
  • 鎖
  • 信號量
  • 事件
  • 條件
  • 定時器
  • 隊(duì)列
  • Python標(biāo)準(zhǔn)模塊--concurrent.futures

操作系統(tǒng)線程理論

線程概念的引入背景

進(jìn)程

　　之前我們已經(jīng)了解了操作系統(tǒng)中進(jìn)程的概念，程序并不能單獨(dú)運(yùn)行，只有將程序裝載到內(nèi)存中，系統(tǒng)為它分配資源才能運(yùn)行，而這種執(zhí)行的程序就稱之為進(jìn)程。程序和進(jìn)程的區(qū)別就在于：程序是指令的集合，它是進(jìn)程運(yùn)行的靜態(tài)描述文本；進(jìn)程是程序的一次執(zhí)行活動，屬于動態(tài)概念。在多道編程中，我們允許多個程序同時加載到內(nèi)存中，在操作系統(tǒng)的調(diào)度下，可以實(shí)現(xiàn)并發(fā)地執(zhí)行。這是這樣的設(shè)計，大大提高了CPU的利用率。進(jìn)程的出現(xiàn)讓每個用戶感覺到自己獨(dú)享CPU，因此，進(jìn)程就是為了在CPU上實(shí)現(xiàn)多道編程而提出的。

有了進(jìn)程為什么要有線程

　　進(jìn)程有很多優(yōu)點(diǎn)，它提供了多道編程，讓我們感覺我們每個人都擁有自己的CPU和其他資源，可以提高計算機(jī)的利用率。很多人就不理解了，既然進(jìn)程這么優(yōu)秀，為什么還要線程呢？其實(shí)，仔細(xì)觀察就會發(fā)現(xiàn)進(jìn)程還是有很多缺陷的，主要體現(xiàn)在兩點(diǎn)上：

• 進(jìn)程只能在一個時間干一件事，如果想同時干兩件事或多件事，進(jìn)程就無能為力了。

• 進(jìn)程在執(zhí)行的過程中如果阻塞，例如等待輸入，整個進(jìn)程就會掛起，即使進(jìn)程中有些工作不依賴于輸入的數(shù)據(jù)，也將無法執(zhí)行。

　　如果這兩個缺點(diǎn)理解比較困難的話，舉個現(xiàn)實(shí)的例子也許你就清楚了：如果把我們上課的過程看成一個進(jìn)程的話，那么我們要做的是耳朵聽老師講課，手上還要記筆記，腦子還要思考問題，這樣才能高效的完成聽課的任務(wù)。而如果只提供進(jìn)程這個機(jī)制的話，上面這三件事將不能同時執(zhí)行，同一時間只能做一件事，聽的時候就不能記筆記，也不能用腦子思考，這是其一；如果老師在黑板上寫演算過程，我們開始記筆記，而老師突然有一步推不下去了，阻塞住了，他在那邊思考著，而我們呢，也不能干其他事，即使你想趁此時思考一下剛才沒聽懂的一個問題都不行，這是其二。

　　現(xiàn)在你應(yīng)該明白了進(jìn)程的缺陷了，而解決的辦法很簡單，我們完全可以讓聽、寫、思三個獨(dú)立的過程，并行起來，這樣很明顯可以提高聽課的效率。而實(shí)際的操作系統(tǒng)中，也同樣引入了這種類似的機(jī)制——線程。

線程的出現(xiàn)

60年代，在OS中能擁有資源和獨(dú)立運(yùn)行的基本單位是進(jìn)程，然而隨著計算機(jī)技術(shù)的發(fā)展，進(jìn)程出現(xiàn)了很多弊端，一是由于進(jìn)程是資源擁有者，創(chuàng)建、撤消與切換存在較大的時空開銷，因此需要引入輕型進(jìn)程；二是由于對稱多處理機(jī)（SMP）出現(xiàn)，可以滿足多個運(yùn)行單位，而多個進(jìn)程并行開銷過大。 因此在80年代，出現(xiàn)了能獨(dú)立運(yùn)行的基本單位——線程（Threads）。 　　注意：進(jìn)程是資源分配的最小單位,線程是CPU調(diào)度的最小單位. 　　　　　每一個進(jìn)程中至少有一個線程。　

進(jìn)程和線程的關(guān)系

　　

?

　　線程與進(jìn)程的區(qū)別可以歸納為以下4點(diǎn)： 1）地址空間和其它資源（如打開文件）：進(jìn)程間相互獨(dú)立，同一進(jìn)程的各線程間共享。某進(jìn)程內(nèi)的線程在其它進(jìn)程不可見。 2）通信：進(jìn)程間通信IPC，線程間可以直接讀寫進(jìn)程數(shù)據(jù)段（如全局變量）來進(jìn)行通信——需要進(jìn)程同步和互斥手段的輔助，以保證數(shù)據(jù)的一致性。 3）調(diào)度和切換：線程上下文切換比進(jìn)程上下文切換要快得多。 4）在多線程操作系統(tǒng)中，進(jìn)程不是一個可執(zhí)行的實(shí)體。 *通過漫畫了解線程進(jìn)城

線程的特點(diǎn)

在多線程的操作系統(tǒng)中，通常是在一個進(jìn)程中包括多個線程，每個線程都是作為利用CPU的基本單位，是花費(fèi)最小開銷的實(shí)體。線程具有以下屬性。 1）輕型實(shí)體 線程中的實(shí)體基本上不擁有系統(tǒng)資源，只是有一點(diǎn)必不可少的、能保證獨(dú)立運(yùn)行的資源。 線程的實(shí)體包括程序、數(shù)據(jù)和TCB。線程是動態(tài)概念，它的動態(tài)特性由線程控制塊TCB（Thread Control Block）描述。 1 TCB包括以下信息： 2 （1）線程狀態(tài)。 3 （2）當(dāng)線程不運(yùn)行時，被保存的現(xiàn)場資源。 4 （3）一組執(zhí)行堆棧。 5 （4）存放每個線程的局部變量主存區(qū)。 6 （5）訪問同一個進(jìn)程中的主存和其它資源。 7 用于指示被執(zhí)行指令序列的程序計數(shù)器、保留局部變量、少數(shù)狀態(tài)參數(shù)和返回地址等的一組寄存器和堆棧。 TCB包括以下信息 2）獨(dú)立調(diào)度和分派的基本單位。 在多線程OS中，線程是能獨(dú)立運(yùn)行的基本單位，因而也是獨(dú)立調(diào)度和分派的基本單位。由于線程很“輕”，故線程的切換非常迅速且開銷小（在同一進(jìn)程中的）。 3）共享進(jìn)程資源。 　　線程在同一進(jìn)程中的各個線程，都可以共享該進(jìn)程所擁有的資源，這首先表現(xiàn)在：所有線程都具有相同的進(jìn)程id，這意味著，線程可以訪問該進(jìn)程的每一個內(nèi)存資源；此外，還可以訪問進(jìn)程所擁有的已打開文件、定時器、信號量機(jī)構(gòu)等。由于同一個進(jìn)程內(nèi)的線程共享內(nèi)存和文件，所以線程之間互相通信不必調(diào)用內(nèi)核。 4）可并發(fā)執(zhí)行。 　　在一個進(jìn)程中的多個線程之間，可以并發(fā)執(zhí)行，甚至允許在一個進(jìn)程中所有線程都能并發(fā)執(zhí)行；同樣，不同進(jìn)程中的線程也能并發(fā)執(zhí)行，充分利用和發(fā)揮了處理機(jī)與外圍設(shè)備并行工作的能力。

使用線程的實(shí)際場景

?

　　開啟一個字處理軟件進(jìn)程，該進(jìn)程肯定需要辦不止一件事情，比如監(jiān)聽鍵盤輸入，處理文字，定時自動將文字保存到硬盤，這三個任務(wù)操作的都是同一塊數(shù)據(jù)，因而不能用多進(jìn)程。只能在一個進(jìn)程里并發(fā)地開啟三個線程,如果是單線程，那就只能是，鍵盤輸入時，不能處理文字和自動保存，自動保存時又不能輸入和處理文字。

內(nèi)存中的線程

?

　　多個線程共享同一個進(jìn)程的地址空間中的資源，是對一臺計算機(jī)上多個進(jìn)程的模擬，有時也稱線程為輕量級的進(jìn)程。

　　而對一臺計算機(jī)上多個進(jìn)程，則共享物理內(nèi)存、磁盤、打印機(jī)等其他物理資源。多線程的運(yùn)行也多進(jìn)程的運(yùn)行類似，是cpu在多個線程之間的快速切換。

　　不同的進(jìn)程之間是充滿敵意的，彼此是搶占、競爭cpu的關(guān)系，如果迅雷會和QQ搶資源。而同一個進(jìn)程是由一個程序員的程序創(chuàng)建，所以同一進(jìn)程內(nèi)的線程是合作關(guān)系，一個線程可以訪問另外一個線程的內(nèi)存地址，大家都是共享的，一個線程干死了另外一個線程的內(nèi)存，那純屬程序員腦子有問題。

　　類似于進(jìn)程，每個線程也有自己的堆棧，不同于進(jìn)程，線程庫無法利用時鐘中斷強(qiáng)制線程讓出CPU，可以調(diào)用thread_yield運(yùn)行線程自動放棄cpu，讓另外一個線程運(yùn)行。

　　線程通常是有益的，但是帶來了不小程序設(shè)計難度，線程的問題是：

　　1. 父進(jìn)程有多個線程，那么開啟的子線程是否需要同樣多的線程

　　2. 在同一個進(jìn)程中，如果一個線程關(guān)閉了文件，而另外一個線程正準(zhǔn)備往該文件內(nèi)寫內(nèi)容呢？

　　因此，在多線程的代碼中，需要更多的心思來設(shè)計程序的邏輯、保護(hù)程序的數(shù)據(jù)。

用戶級線程和內(nèi)核級線程（了解）

　　線程的實(shí)現(xiàn)可以分為兩類：用戶級線程(User-Level Thread)和內(nèi)核線線程(Kernel-Level Thread)，后者又稱為內(nèi)核支持的線程或輕量級進(jìn)程。在多線程操作系統(tǒng)中，各個系統(tǒng)的實(shí)現(xiàn)方式并不相同，在有的系統(tǒng)中實(shí)現(xiàn)了用戶級線程，有的系統(tǒng)中實(shí)現(xiàn)了內(nèi)核級線程。?

用戶級線程

　　內(nèi)核的切換由用戶態(tài)程序自己控制內(nèi)核切換,不需要內(nèi)核干涉，少了進(jìn)出內(nèi)核態(tài)的消耗，但不能很好的利用多核Cpu。

　　

在用戶空間模擬操作系統(tǒng)對進(jìn)程的調(diào)度，來調(diào)用一個進(jìn)程中的線程，每個進(jìn)程中都會有一個運(yùn)行時系統(tǒng)，用來調(diào)度線程。此時當(dāng)該進(jìn)程獲取cpu時，進(jìn)程內(nèi)再調(diào)度出一個線程去執(zhí)行，同一時刻只有一個線程執(zhí)行。

內(nèi)核級線程

　　?內(nèi)核級線程:切換由內(nèi)核控制，當(dāng)線程進(jìn)行切換的時候，由用戶態(tài)轉(zhuǎn)化為內(nèi)核態(tài)。切換完畢要從內(nèi)核態(tài)返回用戶態(tài)；可以很好的利用smp，即利用多核cpu。windows線程就是這樣的。

　　

用戶級與內(nèi)核級線程的對比

1 內(nèi)核支持線程是OS內(nèi)核可感知的，而用戶級線程是OS內(nèi)核不可感知的。 2 用戶級線程的創(chuàng)建、撤消和調(diào)度不需要OS內(nèi)核的支持，是在語言（如Java）這一級處理的；而內(nèi)核支持線程的創(chuàng)建、撤消和調(diào)度都需OS內(nèi)核提供支持，而且與進(jìn)程的創(chuàng)建、撤消和調(diào)度大體是相同的。 3 用戶級線程執(zhí)行系統(tǒng)調(diào)用指令時將導(dǎo)致其所屬進(jìn)程被中斷，而內(nèi)核支持線程執(zhí)行系統(tǒng)調(diào)用指令時，只導(dǎo)致該線程被中斷。 4 在只有用戶級線程的系統(tǒng)內(nèi)，CPU調(diào)度還是以進(jìn)程為單位，處于運(yùn)行狀態(tài)的進(jìn)程中的多個線程，由用戶程序控制線程的輪換運(yùn)行；在有內(nèi)核支持線程的系統(tǒng)內(nèi)，CPU調(diào)度則以線程為單位，由OS的線程調(diào)度程序負(fù)責(zé)線程的調(diào)度。 5 用戶級線程的程序?qū)嶓w是運(yùn)行在用戶態(tài)下的程序，而內(nèi)核支持線程的程序?qū)嶓w則是可以運(yùn)行在任何狀態(tài)下的程序。 用戶級線程和內(nèi)核級線程的區(qū)別 1 優(yōu)點(diǎn)：當(dāng)有多個處理機(jī)時，一個進(jìn)程的多個線程可以同時執(zhí)行。 2 缺點(diǎn)：由內(nèi)核進(jìn)行調(diào)度。 內(nèi)核線程的優(yōu)缺點(diǎn) 1 優(yōu)點(diǎn)： 2 線程的調(diào)度不需要內(nèi)核直接參與，控制簡單。 3 可以在不支持線程的操作系統(tǒng)中實(shí)現(xiàn)。 4 創(chuàng)建和銷毀線程、線程切換代價等線程管理的代價比內(nèi)核線程少得多。 5 允許每個進(jìn)程定制自己的調(diào)度算法，線程管理比較靈活。 6 線程能夠利用的表空間和堆?？臻g比內(nèi)核級線程多。 7 同一進(jìn)程中只能同時有一個線程在運(yùn)行，如果有一個線程使用了系統(tǒng)調(diào)用而阻塞，那么整個進(jìn)程都會被掛起。另外，頁面失效也會產(chǎn)生同樣的問題。 8 缺點(diǎn)： 9 資源調(diào)度按照進(jìn)程進(jìn)行，多個處理機(jī)下，同一個進(jìn)程中的線程只能在同一個處理機(jī)下分時復(fù)用 用戶級線程的優(yōu)缺點(diǎn)

混合實(shí)現(xiàn)

　　用戶級與內(nèi)核級的多路復(fù)用，內(nèi)核同一調(diào)度內(nèi)核線程，每個內(nèi)核線程對應(yīng)n個用戶線程

　　

　　linux操作系統(tǒng)的 NPTL　　　　

1 歷史 2 在內(nèi)核2.6以前的調(diào)度實(shí)體都是進(jìn)程，內(nèi)核并沒有真正支持線程。它是能過一個系統(tǒng)調(diào)用clone()來實(shí)現(xiàn)的，這個調(diào)用創(chuàng)建了一份調(diào)用進(jìn)程的拷貝，跟fork()不同的是,這份進(jìn)程拷貝完全共享了調(diào)用進(jìn)程的地址空間。LinuxThread就是通過這個系統(tǒng)調(diào)用來提供線程在內(nèi)核級的支持的(許多以前的線程實(shí)現(xiàn)都完全是在用戶態(tài)，內(nèi)核根本不知道線程的存在)。非常不幸的是，這種方法有相當(dāng)多的地方?jīng)]有遵循POSIX標(biāo)準(zhǔn)，特別是在信號處理，調(diào)度，進(jìn)程間通信原語等方面。 3 4 很顯然，為了改進(jìn)LinuxThread必須得到內(nèi)核的支持，并且需要重寫線程庫。為了實(shí)現(xiàn)這個需求，開始有兩個相互競爭的項(xiàng)目：IBM啟動的NGTP(Next Generation POSIX Threads)項(xiàng)目，以及Redhat公司的NPTL。在2003年的年中，IBM放棄了NGTP，也就是大約那時，Redhat發(fā)布了最初的NPTL。 5 6 NPTL最開始在redhat linux 9里發(fā)布，現(xiàn)在從RHEL3起內(nèi)核2.6起都支持NPTL，并且完全成了GNU C庫的一部分。 7 8 9 10 設(shè)計 11 NPTL使用了跟LinuxThread相同的辦法，在內(nèi)核里面線程仍然被當(dāng)作是一個進(jìn)程，并且仍然使用了clone()系統(tǒng)調(diào)用(在NPTL庫里調(diào)用)。但是，NPTL需要內(nèi)核級的特殊支持來實(shí)現(xiàn)，比如需要掛起然后再喚醒線程的線程同步原語futex. 12 13 NPTL也是一個1*1的線程庫，就是說，當(dāng)你使用pthread_create()調(diào)用創(chuàng)建一個線程后，在內(nèi)核里就相應(yīng)創(chuàng)建了一個調(diào)度實(shí)體，在linux里就是一個新進(jìn)程，這個方法最大可能的簡化了線程的實(shí)現(xiàn)。 14 15 除NPTL的1*1模型外還有一個m*n模型，通常這種模型的用戶線程數(shù)會比內(nèi)核的調(diào)度實(shí)體多。在這種實(shí)現(xiàn)里，線程庫本身必須去處理可能存在的調(diào)度，這樣在線程庫內(nèi)部的上下文切換通常都會相當(dāng)?shù)目?#xff0c;因?yàn)樗苊饬讼到y(tǒng)調(diào)用轉(zhuǎn)到內(nèi)核態(tài)。然而這種模型增加了線程實(shí)現(xiàn)的復(fù)雜性,并可能出現(xiàn)諸如優(yōu)先級反轉(zhuǎn)的問題，此外，用戶態(tài)的調(diào)度如何跟內(nèi)核態(tài)的調(diào)度進(jìn)行協(xié)調(diào)也是很難讓人滿意。 介紹

線程和python

理論知識

全局解釋器鎖GIL

　　Python代碼的執(zhí)行由Python虛擬機(jī)(也叫解釋器主循環(huán))來控制。Python在設(shè)計之初就考慮到要在主循環(huán)中，同時只有一個線程在執(zhí)行。雖然 Python 解釋器中可以“運(yùn)行”多個線程，但在任意時刻只有一個線程在解釋器中運(yùn)行。
　　對Python虛擬機(jī)的訪問由全局解釋器鎖(GIL)來控制，正是這個鎖能保證同一時刻只有一個線程在運(yùn)行。

　　在多線程環(huán)境中，Python 虛擬機(jī)按以下方式執(zhí)行：

　　a、設(shè)置 GIL；

　　b、切換到一個線程去運(yùn)行；

　　c、運(yùn)行指定數(shù)量的字節(jié)碼指令或者線程主動讓出控制(可以調(diào)用 time.sleep(0))；

　　d、把線程設(shè)置為睡眠狀態(tài)；

　　e、解鎖 GIL；

　　d、再次重復(fù)以上所有步驟。
　　在調(diào)用外部代碼(如 C/C++擴(kuò)展函數(shù))的時候，GIL將會被鎖定，直到這個函數(shù)結(jié)束為止(由于在這期間沒有Python的字節(jié)碼被運(yùn)行，所以不會做線程切換)編寫擴(kuò)展的程序員可以主動解鎖GIL。

python線程模塊的選擇

　　Python提供了幾個用于多線程編程的模塊，包括thread、threading和Queue等。thread和threading模塊允許程序員創(chuàng)建和管理線程。thread模塊提供了基本的線程和鎖的支持，threading提供了更高級別、功能更強(qiáng)的線程管理的功能。Queue模塊允許用戶創(chuàng)建一個可以用于多個線程之間共享數(shù)據(jù)的隊(duì)列數(shù)據(jù)結(jié)構(gòu)。
　　避免使用thread模塊，因?yàn)楦呒墑e的threading模塊更為先進(jìn)，對線程的支持更為完善，而且使用thread模塊里的屬性有可能會與threading出現(xiàn)沖突；其次低級別的thread模塊的同步原語很少(實(shí)際上只有一個)，而threading模塊則有很多；再者，thread模塊中當(dāng)主線程結(jié)束時，所有的線程都會被強(qiáng)制結(jié)束掉，沒有警告也不會有正常的清除工作，至少threading模塊能確保重要的子線程退出后進(jìn)程才退出。?

　　thread模塊不支持守護(hù)線程，當(dāng)主線程退出時，所有的子線程不論它們是否還在工作，都會被強(qiáng)行退出。而threading模塊支持守護(hù)線程，守護(hù)線程一般是一個等待客戶請求的服務(wù)器，如果沒有客戶提出請求它就在那等著，如果設(shè)定一個線程為守護(hù)線程，就表示這個線程是不重要的，在進(jìn)程退出的時候，不用等待這個線程退出。

threading模塊

multiprocess模塊的完全模仿了threading模塊的接口，二者在使用層面，有很大的相似性，因而不再詳細(xì)介紹（官方鏈接）

線程的創(chuàng)建Threading.Thread類

線程的創(chuàng)建

1 from threading import Thread 2 import time 3 def sayhi(name): 4 time.sleep(2) 5 print('%s say hello' %name) 6 7 if __name__ == '__main__': 8 t=Thread(target=sayhi,args=('egon',)) 9 t.start() 10 print('主線程') 創(chuàng)建線程的方式1 1 from threading import Thread 2 import time 3 class Sayhi(Thread): 4 def __init__(self,name): 5 super().__init__() 6 self.name=name 7 def run(self): 8 time.sleep(2) 9 print('%s say hello' % self.name) 10 11 12 if __name__ == '__main__': 13 t = Sayhi('egon') 14 t.start() 15 print('主線程') 創(chuàng)建線程的方式2

多線程與多進(jìn)程

1 from threading import Thread 2 from multiprocessing import Process 3 import os 4 5 def work(): 6 print('hello',os.getpid()) 7 8 if __name__ == '__main__': 9 #part1:在主進(jìn)程下開啟多個線程,每個線程都跟主進(jìn)程的pid一樣 10 t1=Thread(target=work) 11 t2=Thread(target=work) 12 t1.start() 13 t2.start() 14 print('主線程/主進(jìn)程pid',os.getpid()) 15 16 #part2:開多個進(jìn)程,每個進(jìn)程都有不同的pid 17 p1=Process(target=work) 18 p2=Process(target=work) 19 p1.start() 20 p2.start() 21 print('主線程/主進(jìn)程pid',os.getpid()) pid的比較 1 from threading import Thread 2 from multiprocessing import Process 3 import os 4 5 def work(): 6 print('hello') 7 8 if __name__ == '__main__': 9 #在主進(jìn)程下開啟線程 10 t=Thread(target=work) 11 t.start() 12 print('主線程/主進(jìn)程') 13 ''' 14 打印結(jié)果: 15 hello 16 主線程/主進(jìn)程 17 ''' 18 19 #在主進(jìn)程下開啟子進(jìn)程 20 t=Process(target=work) 21 t.start() 22 print('主線程/主進(jìn)程') 23 ''' 24 打印結(jié)果: 25 主線程/主進(jìn)程 26 hello 27 ''' 開啟效率的較量 1 from threading import Thread 2 from multiprocessing import Process 3 import os 4 def work(): 5 global n 6 n=0 7 8 if __name__ == '__main__': 9 # n=100 10 # p=Process(target=work) 11 # p.start() 12 # p.join() 13 # print('主',n) #毫無疑問子進(jìn)程p已經(jīng)將自己的全局的n改成了0,但改的僅僅是它自己的,查看父進(jìn)程的n仍然為100 14 15 16 n=1 17 t=Thread(target=work) 18 t.start() 19 t.join() 20 print('主',n) #查看結(jié)果為0,因?yàn)橥贿M(jìn)程內(nèi)的線程之間共享進(jìn)程內(nèi)的數(shù)據(jù) 21 同一進(jìn)程內(nèi)的線程共享該進(jìn)程的數(shù)據(jù)？ 內(nèi)存數(shù)據(jù)的共享問題

練習(xí) ：多線程實(shí)現(xiàn)socket

1 #_*_coding:utf-8_*_ 2 #!/usr/bin/env python 3 import multiprocessing 4 import threading 5 6 import socket 7 s=socket.socket(socket.AF_INET,socket.SOCK_STREAM) 8 s.bind(('127.0.0.1',8080)) 9 s.listen(5) 10 11 def action(conn): 12 while True: 13 data=conn.recv(1024) 14 print(data) 15 conn.send(data.upper()) 16 17 if __name__ == '__main__': 18 19 while True: 20 conn,addr=s.accept() 21 22 23 p=threading.Thread(target=action,args=(conn,)) 24 p.start() server 1 #_*_coding:utf-8_*_ 2 #!/usr/bin/env python 3 4 5 import socket 6 7 s=socket.socket(socket.AF_INET,socket.SOCK_STREAM) 8 s.connect(('127.0.0.1',8080)) 9 10 while True: 11 msg=input('>>: ').strip() 12 if not msg:continue 13 14 s.send(msg.encode('utf-8')) 15 data=s.recv(1024) 16 print(data) client

Thread類的其他方法

Thread實(shí)例對象的方法# isAlive(): 返回線程是否活動的。# getName(): 返回線程名。# setName(): 設(shè)置線程名。 threading模塊提供的一些方法：# threading.currentThread(): 返回當(dāng)前的線程變量。# threading.enumerate(): 返回一個包含正在運(yùn)行的線程的list。正在運(yùn)行指線程啟動后、結(jié)束前，不包括啟動前和終止后的線程。# threading.activeCount(): 返回正在運(yùn)行的線程數(shù)量，與len(threading.enumerate())有相同的結(jié)果。 1 from threading import Thread 2 import threading 3 from multiprocessing import Process 4 import os 5 6 def work(): 7 import time 8 time.sleep(3) 9 print(threading.current_thread().getName()) 10 11 12 if __name__ == '__main__': 13 #在主進(jìn)程下開啟線程 14 t=Thread(target=work) 15 t.start() 16 17 print(threading.current_thread().getName()) 18 print(threading.current_thread()) #主線程 19 print(threading.enumerate()) #連同主線程在內(nèi)有兩個運(yùn)行的線程 20 print(threading.active_count()) 21 print('主線程/主進(jìn)程') 22 23 ''' 24 打印結(jié)果: 25 MainThread 26 <_MainThread(MainThread, started 140735268892672)> 27 [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>] 28 主線程/主進(jìn)程 29 Thread-1 30 ''' 代碼示例 1 from threading import Thread 2 import time 3 def sayhi(name): 4 time.sleep(2) 5 print('%s say hello' %name) 6 7 if __name__ == '__main__': 8 t=Thread(target=sayhi,args=('egon',)) 9 t.start() 10 t.join() 11 print('主線程') 12 print(t.is_alive()) 13 ''' 14 egon say hello 15 主線程 16 False 17 ''' join方法

守護(hù)線程

無論是進(jìn)程還是線程，都遵循：守護(hù)xx會等待主xx運(yùn)行完畢后被銷毀。需要強(qiáng)調(diào)的是：運(yùn)行完畢并非終止運(yùn)行

#1.對主進(jìn)程來說，運(yùn)行完畢指的是主進(jìn)程代碼運(yùn)行完畢 #2.對主線程來說，運(yùn)行完畢指的是主線程所在的進(jìn)程內(nèi)所有非守護(hù)線程統(tǒng)統(tǒng)運(yùn)行完畢，主線程才算運(yùn)行完畢 1 #1 主進(jìn)程在其代碼結(jié)束后就已經(jīng)算運(yùn)行完畢了（守護(hù)進(jìn)程在此時就被回收）,然后主進(jìn)程會一直等非守護(hù)的子進(jìn)程都運(yùn)行完畢后回收子進(jìn)程的資源(否則會產(chǎn)生僵尸進(jìn)程)，才會結(jié)束， 2 #2 主線程在其他非守護(hù)線程運(yùn)行完畢后才算運(yùn)行完畢（守護(hù)線程在此時就被回收）。因?yàn)橹骶€程的結(jié)束意味著進(jìn)程的結(jié)束，進(jìn)程整體的資源都將被回收，而進(jìn)程必須保證非守護(hù)線程都運(yùn)行完畢后才能結(jié)束。 詳細(xì)解釋 1 from threading import Thread 2 import time 3 def sayhi(name): 4 time.sleep(2) 5 print('%s say hello' %name) 6 7 if __name__ == '__main__': 8 t=Thread(target=sayhi,args=('egon',)) 9 t.setDaemon(True) #必須在t.start()之前設(shè)置 10 t.start() 11 12 print('主線程') 13 print(t.is_alive()) 14 ''' 15 主線程 16 True 17 ''' 守護(hù)線程例1 1 from threading import Thread 2 import time 3 def foo(): 4 print(123) 5 time.sleep(1) 6 print("end123") 7 8 def bar(): 9 print(456) 10 time.sleep(3) 11 print("end456") 12 13 14 t1=Thread(target=foo) 15 t2=Thread(target=bar) 16 17 t1.daemon=True 18 t1.start() 19 t2.start() 20 print("main-------") 守護(hù)線程例2

鎖

鎖與GIL

?

同步鎖

1 from threading import Thread 2 import os,time 3 def work(): 4 global n 5 temp=n 6 time.sleep(0.1) 7 n=temp-1 8 if __name__ == '__main__': 9 n=100 10 l=[] 11 for i in range(100): 12 p=Thread(target=work) 13 l.append(p) 14 p.start() 15 for p in l: 16 p.join() 17 18 print(n) #結(jié)果可能為99 多個線程搶占資源的情況 import threading R=threading.Lock() R.acquire() ''' 對公共數(shù)據(jù)的操作 ''' R.release()

　　

1 from threading import Thread,Lock 2 import os,time 3 def work(): 4 global n 5 lock.acquire() 6 temp=n 7 time.sleep(0.1) 8 n=temp-1 9 lock.release() 10 if __name__ == '__main__': 11 lock=Lock() 12 n=100 13 l=[] 14 for i in range(100): 15 p=Thread(target=work) 16 l.append(p) 17 p.start() 18 for p in l: 19 p.join() 20 21 print(n) #結(jié)果肯定為0，由原來的并發(fā)執(zhí)行變成串行，犧牲了執(zhí)行效率保證了數(shù)據(jù)安全 同步鎖的引用 1 #不加鎖:并發(fā)執(zhí)行,速度快,數(shù)據(jù)不安全 2 from threading import current_thread,Thread,Lock 3 import os,time 4 def task(): 5 global n 6 print('%s is running' %current_thread().getName()) 7 temp=n 8 time.sleep(0.5) 9 n=temp-1 10 11 12 if __name__ == '__main__': 13 n=100 14 lock=Lock() 15 threads=[] 16 start_time=time.time() 17 for i in range(100): 18 t=Thread(target=task) 19 threads.append(t) 20 t.start() 21 for t in threads: 22 t.join() 23 24 stop_time=time.time() 25 print('主:%s n:%s' %(stop_time-start_time,n)) 26 27 ''' 28 Thread-1 is running 29 Thread-2 is running 30 ...... 31 Thread-100 is running 32 主:0.5216062068939209 n:99 33 ''' 34 35 36 #不加鎖:未加鎖部分并發(fā)執(zhí)行,加鎖部分串行執(zhí)行,速度慢,數(shù)據(jù)安全 37 from threading import current_thread,Thread,Lock 38 import os,time 39 def task(): 40 #未加鎖的代碼并發(fā)運(yùn)行 41 time.sleep(3) 42 print('%s start to run' %current_thread().getName()) 43 global n 44 #加鎖的代碼串行運(yùn)行 45 lock.acquire() 46 temp=n 47 time.sleep(0.5) 48 n=temp-1 49 lock.release() 50 51 if __name__ == '__main__': 52 n=100 53 lock=Lock() 54 threads=[] 55 start_time=time.time() 56 for i in range(100): 57 t=Thread(target=task) 58 threads.append(t) 59 t.start() 60 for t in threads: 61 t.join() 62 stop_time=time.time() 63 print('主:%s n:%s' %(stop_time-start_time,n)) 64 65 ''' 66 Thread-1 is running 67 Thread-2 is running 68 ...... 69 Thread-100 is running 70 主:53.294203758239746 n:0 71 ''' 72 73 #有的同學(xué)可能有疑問:既然加鎖會讓運(yùn)行變成串行,那么我在start之后立即使用join,就不用加鎖了啊,也是串行的效果啊 74 #沒錯:在start之后立刻使用jion,肯定會將100個任務(wù)的執(zhí)行變成串行,毫無疑問,最終n的結(jié)果也肯定是0,是安全的,但問題是 75 #start后立即join:任務(wù)內(nèi)的所有代碼都是串行執(zhí)行的,而加鎖,只是加鎖的部分即修改共享數(shù)據(jù)的部分是串行的 76 #單從保證數(shù)據(jù)安全方面,二者都可以實(shí)現(xiàn),但很明顯是加鎖的效率更高. 77 from threading import current_thread,Thread,Lock 78 import os,time 79 def task(): 80 time.sleep(3) 81 print('%s start to run' %current_thread().getName()) 82 global n 83 temp=n 84 time.sleep(0.5) 85 n=temp-1 86 87 88 if __name__ == '__main__': 89 n=100 90 lock=Lock() 91 start_time=time.time() 92 for i in range(100): 93 t=Thread(target=task) 94 t.start() 95 t.join() 96 stop_time=time.time() 97 print('主:%s n:%s' %(stop_time-start_time,n)) 98 99 ''' 100 Thread-1 start to run 101 Thread-2 start to run 102 ...... 103 Thread-100 start to run 104 主:350.6937336921692 n:0 #耗時是多么的恐怖 105 ''' 106 107 ） 互斥鎖與join的區(qū)別

?

死鎖與遞歸鎖

進(jìn)程也有死鎖與遞歸鎖，在進(jìn)程那里忘記說了，放到這里一切說了額

所謂死鎖： 是指兩個或兩個以上的進(jìn)程或線程在執(zhí)行過程中，因爭奪資源而造成的一種互相等待的現(xiàn)象，若無外力作用，它們都將無法推進(jìn)下去。此時稱系統(tǒng)處于死鎖狀態(tài)或系統(tǒng)產(chǎn)生了死鎖，這些永遠(yuǎn)在互相等待的進(jìn)程稱為死鎖進(jìn)程，如下就是死鎖

1 from threading import Lock as Lock 2 import time 3 mutexA=Lock() 4 mutexA.acquire() 5 mutexA.acquire() 6 print(123) 7 mutexA.release() 8 mutexA.release() 死鎖

解決方法，遞歸鎖，在Python中為了支持在同一線程中多次請求同一資源，python提供了可重入鎖RLock。

這個RLock內(nèi)部維護(hù)著一個Lock和一個counter變量，counter記錄了acquire的次數(shù)，從而使得資源可以被多次require。直到一個線程所有的acquire都被release，其他的線程才能獲得資源。上面的例子如果使用RLock代替Lock，則不會發(fā)生死鎖：

1 from threading import RLock as Lock 2 import time 3 mutexA=Lock() 4 mutexA.acquire() 5 mutexA.acquire() 6 print(123) 7 mutexA.release() 8 mutexA.release() 遞歸鎖RLock

典型問題：科學(xué)家吃面

1 import time 2 from threading import Thread,Lock 3 noodle_lock = Lock() 4 fork_lock = Lock() 5 def eat1(name): 6 noodle_lock.acquire() 7 print('%s 搶到了面條'%name) 8 fork_lock.acquire() 9 print('%s 搶到了叉子'%name) 10 print('%s 吃面'%name) 11 fork_lock.release() 12 noodle_lock.release() 13 14 def eat2(name): 15 fork_lock.acquire() 16 print('%s 搶到了叉子' % name) 17 time.sleep(1) 18 noodle_lock.acquire() 19 print('%s 搶到了面條' % name) 20 print('%s 吃面' % name) 21 noodle_lock.release() 22 fork_lock.release() 23 24 for name in ['哪吒','egon','yuan']: 25 t1 = Thread(target=eat1,args=(name,)) 26 t2 = Thread(target=eat2,args=(name,)) 27 t1.start() 28 t2.start() 死鎖問題 1 import time 2 from threading import Thread,RLock 3 fork_lock = noodle_lock = RLock() 4 def eat1(name): 5 noodle_lock.acquire() 6 print('%s 搶到了面條'%name) 7 fork_lock.acquire() 8 print('%s 搶到了叉子'%name) 9 print('%s 吃面'%name) 10 fork_lock.release() 11 noodle_lock.release() 12 13 def eat2(name): 14 fork_lock.acquire() 15 print('%s 搶到了叉子' % name) 16 time.sleep(1) 17 noodle_lock.acquire() 18 print('%s 搶到了面條' % name) 19 print('%s 吃面' % name) 20 noodle_lock.release() 21 fork_lock.release() 22 23 for name in ['哪吒','egon','yuan']: 24 t1 = Thread(target=eat1,args=(name,)) 25 t2 = Thread(target=eat2,args=(name,)) 26 t1.start() 27 t2.start() 遞歸鎖解決死鎖問題

信號量

同進(jìn)程的一樣

Semaphore管理一個內(nèi)置的計數(shù)器，
每當(dāng)調(diào)用acquire()時內(nèi)置計數(shù)器-1；
調(diào)用release() 時內(nèi)置計數(shù)器+1；
計數(shù)器不能小于0；當(dāng)計數(shù)器為0時，acquire()將阻塞線程直到其他線程調(diào)用release()。

實(shí)例：(同時只有5個線程可以獲得semaphore,即可以限制最大連接數(shù)為5)：

1 from threading import Thread,Semaphore 2 import threading 3 import time 4 # def func(): 5 # if sm.acquire(): 6 # print (threading.currentThread().getName() + ' get semaphore') 7 # time.sleep(2) 8 # sm.release() 9 def func(): 10 sm.acquire() 11 print('%s get sm' %threading.current_thread().getName()) 12 time.sleep(3) 13 sm.release() 14 if __name__ == '__main__': 15 sm=Semaphore(5) 16 for i in range(23): 17 t=Thread(target=func) 18 t.start() 實(shí)例 與進(jìn)程池是完全不同的概念，進(jìn)程池Pool(4)，最大只能產(chǎn)生4個進(jìn)程，而且從頭到尾都只是這四個進(jìn)程，不會產(chǎn)生新的，而信號量是產(chǎn)生一堆線程/進(jìn)程 池與信號量

事件

同進(jìn)程的一樣

線程的一個關(guān)鍵特性是每個線程都是獨(dú)立運(yùn)行且狀態(tài)不可預(yù)測。如果程序中的其 他線程需要通過判斷某個線程的狀態(tài)來確定自己下一步的操作,這時線程同步問題就會變得非常棘手。為了解決這些問題,我們需要使用threading庫中的Event對象。 對象包含一個可由線程設(shè)置的信號標(biāo)志,它允許線程等待某些事件的發(fā)生。在 初始情況下,Event對象中的信號標(biāo)志被設(shè)置為假。如果有線程等待一個Event對象, 而這個Event對象的標(biāo)志為假,那么這個線程將會被一直阻塞直至該標(biāo)志為真。一個線程如果將一個Event對象的信號標(biāo)志設(shè)置為真,它將喚醒所有等待這個Event對象的線程。如果一個線程等待一個已經(jīng)被設(shè)置為真的Event對象,那么它將忽略這個事件, 繼續(xù)執(zhí)行

event.isSet()：返回event的狀態(tài)值； event.wait()：如果 event.isSet()==False將阻塞線程； event.set()： 設(shè)置event的狀態(tài)值為True，所有阻塞池的線程激活進(jìn)入就緒狀態(tài)， 等待操作系統(tǒng)調(diào)度； event.clear()：恢復(fù)event的狀態(tài)值為False。

　　????

例如，有多個工作線程嘗試鏈接MySQL，我們想要在鏈接前確保MySQL服務(wù)正常才讓那些工作線程去連接MySQL服務(wù)器，如果連接不成功，都會去嘗試重新連接。那么我們就可以采用threading.Event機(jī)制來協(xié)調(diào)各個工作線程的連接操作

1 import threading 2 import time,random 3 from threading import Thread,Event 4 5 def conn_mysql(): 6 count=1 7 while not event.is_set(): 8 if count > 3: 9 raise TimeoutError('鏈接超時') 10 print('<%s>第%s次嘗試鏈接' % (threading.current_thread().getName(), count)) 11 event.wait(0.5) 12 count+=1 13 print('<%s>鏈接成功' %threading.current_thread().getName()) 14 15 16 def check_mysql(): 17 print('\033[45m[%s]正在檢查mysql\033[0m' % threading.current_thread().getName()) 18 time.sleep(random.randint(2,4)) 19 event.set() 20 if __name__ == '__main__': 21 event=Event() 22 conn1=Thread(target=conn_mysql) 23 conn2=Thread(target=conn_mysql) 24 check=Thread(target=check_mysql) 25 26 conn1.start() 27 conn2.start() 28 check.start() 實(shí)例

條件

使得線程等待，只有滿足某條件時，才釋放n個線程

1 Python提供的Condition對象提供了對復(fù)雜線程同步問題的支持。Condition被稱為條件變量，除了提供與Lock類似的acquire和release方法外，還提供了wait和notify方法。線程首先acquire一個條件變量，然后判斷一些條件。如果條件不滿足則wait；如果條件滿足，進(jìn)行一些處理改變條件后，通過notify方法通知其他線程，其他處于wait狀態(tài)的線程接到通知后會重新判斷條件。不斷的重復(fù)這一過程，從而解決復(fù)雜的同步問題。 詳細(xì)說明 1 import threading 2 3 def run(n): 4 con.acquire() 5 con.wait() 6 print("run the thread: %s" % n) 7 con.release() 8 9 if __name__ == '__main__': 10 11 con = threading.Condition() 12 for i in range(10): 13 t = threading.Thread(target=run, args=(i,)) 14 t.start() 15 16 while True: 17 inp = input('>>>') 18 if inp == 'q': 19 break 20 con.acquire() 21 con.notify(int(inp)) 22 con.release() 23 print('****') 實(shí)例

定時器

定時器，指定n秒后執(zhí)行某個操作

1 from threading import Timer 2 3 def hello(): 4 print("hello, world") 5 6 t = Timer(1, hello) 7 t.start() # after 1 seconds, "hello, world" will be printed View Code

線程隊(duì)列

queue隊(duì)列 ：使用import queue，用法與進(jìn)程Queue一樣

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

class?queue.Queue(maxsize=0) #先進(jìn)先出 1 import queue 2 3 q=queue.Queue() 4 q.put('first') 5 q.put('second') 6 q.put('third') 7 8 print(q.get()) 9 print(q.get()) 10 print(q.get()) 11 ''' 12 結(jié)果(先進(jìn)先出): 13 first 14 second 15 third 16 ''' 先進(jìn)先出

class?queue.LifoQueue(maxsize=0) #last in fisrt out

1 import queue 2 3 q=queue.LifoQueue() 4 q.put('first') 5 q.put('second') 6 q.put('third') 7 8 print(q.get()) 9 print(q.get()) 10 print(q.get()) 11 ''' 12 結(jié)果(后進(jìn)先出): 13 third 14 second 15 first 16 ''' 后進(jìn)先出

class?queue.PriorityQueue(maxsize=0) #存儲數(shù)據(jù)時可設(shè)置優(yōu)先級的隊(duì)列

1 import queue 2 3 q=queue.PriorityQueue() 4 #put進(jìn)入一個元組,元組的第一個元素是優(yōu)先級(通常是數(shù)字,也可以是非數(shù)字之間的比較),數(shù)字越小優(yōu)先級越高 5 q.put((20,'a')) 6 q.put((10,'b')) 7 q.put((30,'c')) 8 9 print(q.get()) 10 print(q.get()) 11 print(q.get()) 12 ''' 13 結(jié)果(數(shù)字越小優(yōu)先級越高,優(yōu)先級高的優(yōu)先出隊(duì)): 14 (10, 'b') 15 (20, 'a') 16 (30, 'c') 17 ''' 優(yōu)先級隊(duì)列 1 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. 2 3 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). 4 5 exception queue.Empty 6 Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty. 7 8 exception queue.Full 9 Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full. 10 11 Queue.qsize() 12 Queue.empty() #return True if empty 13 Queue.full() # return True if full 14 Queue.put(item, block=True, timeout=None) 15 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). 16 17 Queue.put_nowait(item) 18 Equivalent to put(item, False). 19 20 Queue.get(block=True, timeout=None) 21 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). 22 23 Queue.get_nowait() 24 Equivalent to get(False). 25 26 Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads. 27 28 Queue.task_done() 29 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. 30 31 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). 32 33 Raises a ValueError if called more times than there were items placed in the queue. 34 35 Queue.join() block直到queue被消費(fèi)完畢 更多方法說明

Python標(biāo)準(zhǔn)模塊--concurrent.futures

https://docs.python.org/dev/library/concurrent.futures.html

#1 介紹 concurrent.futures模塊提供了高度封裝的異步調(diào)用接口 ThreadPoolExecutor：線程池，提供異步調(diào)用 ProcessPoolExecutor: 進(jìn)程池，提供異步調(diào)用 Both implement the same interface, which is defined by the abstract Executor class.#2 基本方法 #submit(fn, *args, **kwargs) 異步提交任務(wù)#map(func, *iterables, timeout=None, chunksize=1) 取代for循環(huán)submit的操作#shutdown(wait=True) 相當(dāng)于進(jìn)程池的pool.close()+pool.join()操作 wait=True，等待池內(nèi)所有任務(wù)執(zhí)行完畢回收完資源后才繼續(xù) wait=False，立即返回，并不會等待池內(nèi)的任務(wù)執(zhí)行完畢 但不管wait參數(shù)為何值，整個程序都會等到所有任務(wù)執(zhí)行完畢 submit和map必須在shutdown之前#result(timeout=None) 取得結(jié)果#add_done_callback(fn) 回調(diào)函數(shù) 1 #介紹 2 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. 3 4 class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None) 5 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. 6 7 8 #用法 9 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor 10 11 import os,time,random 12 def task(n): 13 print('%s is runing' %os.getpid()) 14 time.sleep(random.randint(1,3)) 15 return n**2 16 17 if __name__ == '__main__': 18 19 executor=ProcessPoolExecutor(max_workers=3) 20 21 futures=[] 22 for i in range(11): 23 future=executor.submit(task,i) 24 futures.append(future) 25 executor.shutdown(True) 26 print('+++>') 27 for future in futures: 28 print(future.result()) ProcessPoolExecutor 1 #介紹 2 ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously. 3 class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='') 4 An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously. 5 6 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. 7 8 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. 9 10 #用法 11 與ProcessPoolExecutor相同 ThreadPoolExecutor 1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor 2 3 import os,time,random 4 def task(n): 5 print('%s is runing' %os.getpid()) 6 time.sleep(random.randint(1,3)) 7 return n**2 8 9 if __name__ == '__main__': 10 11 executor=ThreadPoolExecutor(max_workers=3) 12 13 # for i in range(11): 14 # future=executor.submit(task,i) 15 16 executor.map(task,range(1,12)) #map取代了for+submit map的用法 1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor 2 from multiprocessing import Pool 3 import requests 4 import json 5 import os 6 7 def get_page(url): 8 print('<進(jìn)程%s> get %s' %(os.getpid(),url)) 9 respone=requests.get(url) 10 if respone.status_code == 200: 11 return {'url':url,'text':respone.text} 12 13 def parse_page(res): 14 res=res.result() 15 print('<進(jìn)程%s> parse %s' %(os.getpid(),res['url'])) 16 parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text'])) 17 with open('db.txt','a') as f: 18 f.write(parse_res) 19 20 21 if __name__ == '__main__': 22 urls=[ 23 'https://www.baidu.com', 24 'https://www.python.org', 25 'https://www.openstack.org', 26 'https://help.github.com/', 27 'http://www.sina.com.cn/' 28 ] 29 30 # p=Pool(3) 31 # for url in urls: 32 # p.apply_async(get_page,args=(url,),callback=pasrse_page) 33 # p.close() 34 # p.join() 35 36 p=ProcessPoolExecutor(3) 37 for url in urls: 38 p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一個future對象obj，需要用obj.result()拿到結(jié)果 回調(diào)函數(shù)

?

?

?

轉(zhuǎn)載于:https://www.cnblogs.com/liuye1990/p/9379795.html

總結(jié)

以上是生活随笔為你收集整理的python - 线程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。