python之路——線程

簡(jiǎn)介

• 操作系統(tǒng)線程理論
  • 線程概念的引入背景
  • 線程的特點(diǎn)
  • 進(jìn)程和線程的關(guān)系
  • 使用線程的實(shí)際場(chǎng)景
  • 用戶級(jí)線程和內(nèi)核級(jí)線程(了解)
• 線程和python
  • 理論知識(shí)
  • 線程的創(chuàng)建Threading.Thread類
  • 鎖
  • 信號(hào)量
  • 事件
  • 條件
  • 定時(shí)器
  • 隊(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í)行活動(dòng)，屬于動(dòng)態(tài)概念。在多道編程中，我們?cè)试S多個(gè)程序同時(shí)加載到內(nèi)存中，在操作系統(tǒng)的調(diào)度下，可以實(shí)現(xiàn)并發(fā)地執(zhí)行。這是這樣的設(shè)計(jì)，大大提高了CPU的利用率。進(jìn)程的出現(xiàn)讓每個(gè)用戶感覺(jué)到自己獨(dú)享CPU，因此，進(jìn)程就是為了在CPU上實(shí)現(xiàn)多道編程而提出的。

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

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

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

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

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

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

線程的出現(xiàn)

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

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

　　

?

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

線程的特點(diǎn)

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

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

?

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

內(nèi)存中的線程

?

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

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

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

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

　　線程通常是有益的，但是帶來(lái)了不小程序設(shè)計(jì)難度，線程的問(wèn)題是：

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

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

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

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

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

用戶級(jí)線程

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

　　

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

內(nèi)核級(jí)線程

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

　　

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

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

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

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

　　

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

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

線程和python

理論知識(shí)

全局解釋器鎖GIL

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

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

　　a、設(shè)置 GIL；

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

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

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

　　e、解鎖 GIL；

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

python線程模塊的選擇

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

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

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)程下開(kāi)啟多個(gè)線程,每個(gè)線程都跟主進(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:開(kāi)多個(gè)進(jìn)程,每個(gè)進(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)程下開(kāi)啟線程 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)程下開(kāi)啟子進(jìn)程 20 t=Process(target=work) 21 t.start() 22 print('主線程/主進(jìn)程') 23 ''' 24 打印結(jié)果: 25 主線程/主進(jìn)程 26 hello 27 ''' 開(kāi)啟效率的較量 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) #毫無(wú)疑問(wè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ù)的共享問(wèn)題

練習(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í)例對(duì)象的方法# isAlive(): 返回線程是否活動(dòng)的。# getName(): 返回線程名。# setName(): 設(shè)置線程名。 threading模塊提供的一些方法：# threading.currentThread(): 返回當(dāng)前的線程變量。# threading.enumerate(): 返回一個(gè)包含正在運(yùn)行的線程的list。正在運(yùn)行指線程啟動(dòng)后、結(jié)束前，不包括啟動(dòng)前和終止后的線程。# 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)程下開(kāi)啟線程 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)有兩個(gè)運(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ù)線程

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

#1.對(duì)主進(jìn)程來(lái)說(shuō)，運(yùn)行完畢指的是主進(jìn)程代碼運(yùn)行完畢 #2.對(duì)主線程來(lái)說(shuō)，運(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)程在此時(shí)就被回收）,然后主進(jìn)程會(huì)一直等非守護(hù)的子進(jìn)程都運(yùn)行完畢后回收子進(jìn)程的資源(否則會(huì)產(chǎn)生僵尸進(jìn)程)，才會(huì)結(jié)束， 2 #2 主線程在其他非守護(hù)線程運(yùn)行完畢后才算運(yùn)行完畢（守護(hù)線程在此時(shí)就被回收）。因?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 多個(gè)線程搶占資源的情況 import threading R=threading.Lock() R.acquire() ''' 對(duì)公共數(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，由原來(lái)的并發(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é)可能有疑問(wèn):既然加鎖會(huì)讓運(yùn)行變成串行,那么我在start之后立即使用join,就不用加鎖了啊,也是串行的效果啊 74 #沒(méi)錯(cuò):在start之后立刻使用jion,肯定會(huì)將100個(gè)任務(wù)的執(zhí)行變成串行,毫無(wú)疑問(wèn),最終n的結(jié)果也肯定是0,是安全的,但問(wèn)題是 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 #耗時(shí)是多么的恐怖 105 ''' 106 107 ） 互斥鎖與join的區(qū)別

?

死鎖與遞歸鎖

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

所謂死鎖： 是指兩個(gè)或兩個(gè)以上的進(jìn)程或線程在執(zhí)行過(guò)程中，因爭(zhēng)奪資源而造成的一種互相等待的現(xiàn)象，若無(wú)外力作用，它們都將無(wú)法推進(jìn)下去。此時(shí)稱系統(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中為了支持在同一線程中多次請(qǐng)求同一資源，python提供了可重入鎖RLock。

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

典型問(wèn)題：科學(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() 死鎖問(wèn)題 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() 遞歸鎖解決死鎖問(wèn)題

信號(hào)量

同進(jìn)程的一樣

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

實(shí)例：(同時(shí)只有5個(gè)線程可以獲得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個(gè)進(jìn)程，而且從頭到尾都只是這四個(gè)進(jìn)程，不會(huì)產(chǎn)生新的，而信號(hào)量是產(chǎn)生一堆線程/進(jìn)程 池與信號(hào)量

事件

同進(jìn)程的一樣

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

　　????

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

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('鏈接超時(shí)') 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í)例

條件

使得線程等待，只有滿足某條件時(shí)，才釋放n個(gè)線程

1 Python提供的Condition對(duì)象提供了對(duì)復(fù)雜線程同步問(wèn)題的支持。Condition被稱為條件變量，除了提供與Lock類似的acquire和release方法外，還提供了wait和notify方法。線程首先acquire一個(gè)條件變量，然后判斷一些條件。如果條件不滿足則wait；如果條件滿足，進(jìn)行一些處理改變條件后，通過(guò)notify方法通知其他線程，其他處于wait狀態(tài)的線程接到通知后會(huì)重新判斷條件。不斷的重復(fù)這一過(guò)程，從而解決復(fù)雜的同步問(wèn)題。 詳細(xì)說(shuō)明 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í)例

定時(shí)器

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

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) #存儲(chǔ)數(shù)據(jù)時(shí)可設(shè)置優(yōu)先級(jí)的隊(duì)列

1 import queue 2 3 q=queue.PriorityQueue() 4 #put進(jìn)入一個(gè)元組,元組的第一個(gè)元素是優(yōu)先級(jí)(通常是數(shù)字,也可以是非數(shù)字之間的比較),數(shù)字越小優(yōu)先級(jí)越高 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)先級(jí)越高,優(yōu)先級(jí)高的優(yōu)先出隊(duì)): 14 (10, 'b') 15 (20, 'a') 16 (30, 'c') 17 ''' 優(yōu)先級(jí)隊(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)完畢 更多方法說(shuō)明

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，立即返回，并不會(huì)等待池內(nèi)的任務(wù)執(zhí)行完畢 但不管wait參數(shù)為何值，整個(gè)程序都會(huì)等到所有任務(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拿到的是一個(gè)future對(duì)象obj，需要用obj.result()拿到結(jié)果 回調(diào)函數(shù)

?

?

?

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

總結(jié)

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

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