線程相關

時間 2019-12-05

標籤線程相關简体版

原文原文鏈接

操做系統線程理論

線程概念的引入背景

進程

　　以前咱們已經瞭解了操做系統中進程的概念，程序並不能單獨運行，只有將程序裝載到內存中，系統爲它分配資源才能運行，而這種執行的程序就稱之爲進程。程序和進程的區別就在於：程序是指令的集合，它是進程運行的靜態描述文本；進程是程序的一次執行活動，屬於動態概念。在多道編程中，咱們容許多個程序同時加載到內存中，在操做系統的調度下，能夠實現併發地執行。這是這樣的設計，大大提升了CPU的利用率。進程的出現讓每一個用戶感受到本身獨享CPU，所以，進程就是爲了在CPU上實現多道編程而提出的。html

有了進程爲何要有線程

　　進程有不少優勢，它提供了多道編程，讓咱們感受咱們每一個人都擁有本身的CPU和其餘資源，能夠提升計算機的利用率。不少人就不理解了，既然進程這麼優秀，爲何還要線程呢？其實，仔細觀察就會發現進程仍是有不少缺陷的，主要體如今兩點上：python

進程只能在一個時間幹一件事，若是想同時幹兩件事或多件事，進程就無能爲力了。linux
進程在執行的過程當中若是阻塞，例如等待輸入，整個進程就會掛起，即便進程中有些工做不依賴於輸入的數據，也將沒法執行。git

　　若是這兩個缺點理解比較困難的話，舉個現實的例子也許你就清楚了：若是把咱們上課的過程當作一個進程的話，那麼咱們要作的是耳朵聽老師講課，手上還要記筆記，腦子還要思考問題，這樣才能高效的完成聽課的任務。而若是隻提供進程這個機制的話，上面這三件事將不能同時執行，同一時間只能作一件事，聽的時候就不能記筆記，也不能用腦子思考，這是其一；若是老師在黑板上寫演算過程，咱們開始記筆記，而老師忽然有一步推不下去了，阻塞住了，他在那邊思考着，而咱們呢，也不能幹其餘事，即便你想趁此時思考一下剛纔沒聽懂的一個問題都不行，這是其二。程序員

　　如今你應該明白了進程的缺陷了，而解決的辦法很簡單，咱們徹底可讓聽、寫、思三個獨立的過程，並行起來，這樣很明顯能夠提升聽課的效率。而實際的操做系統中，也一樣引入了這種相似的機制——線程。github

線程的出現

　　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在多個線程之間的快速切換。json

　　不一樣的進程之間是充滿敵意的，彼此是搶佔、競爭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模塊的接口，兩者在使用層面，有很大的類似性，於是再也不詳細介紹（官方連接）

回到頂部

線程的建立Threading.Thread類

線程的建立

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    print('主線程')

建立線程的方式1

from threading import Thread
import time
class Sayhi(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        time.sleep(2)
        print('%s say hello' % self.name)


if __name__ == '__main__':
    t = Sayhi('egon')
    t.start()
    print('主線程')

建立線程的方式2

多線程與多進程

from threading import Thread
from multiprocessing import Process
import os

def work():
    print('hello',os.getpid())

if __name__ == '__main__':
    #part1:在主進程下開啓多個線程,每一個線程都跟主進程的pid同樣
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print('主線程/主進程pid',os.getpid())

    #part2:開多個進程,每一個進程都有不一樣的pid
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print('主線程/主進程pid',os.getpid())

pid的比較

from threading import Thread
from multiprocessing import Process
import os

def work():
    print('hello')

if __name__ == '__main__':
    #在主進程下開啓線程
    t=Thread(target=work)
    t.start()
    print('主線程/主進程')
    '''
    打印結果:
    hello
    主線程/主進程
    '''

    #在主進程下開啓子進程
    t=Process(target=work)
    t.start()
    print('主線程/主進程')
    '''
    打印結果:
    主線程/主進程
    hello
    '''

開啓效率的較量

from  threading import Thread
from multiprocessing import Process
import os
def work():
    global n
    n=0

if __name__ == '__main__':
    # n=100
    # p=Process(target=work)
    # p.start()
    # p.join()
    # print('主',n) #毫無疑問子進程p已經將本身的全局的n改爲了0,但改的僅僅是它本身的,查看父進程的n仍然爲100


    n=1
    t=Thread(target=work)
    t.start()
    t.join()
    print('主',n) #查看結果爲0,由於同一進程內的線程之間共享進程內的數據
同一進程內的線程共享該進程的數據？

內存數據的共享問題

練習：多線程實現socket

#_*_coding:utf-8_*_
#!/usr/bin/env python
import multiprocessing
import threading

import socket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('127.0.0.1',8080))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print(data)
        conn.send(data.upper())

if __name__ == '__main__':

    while True:
        conn,addr=s.accept()


        p=threading.Thread(target=action,args=(conn,))
        p.start()

server

#_*_coding:utf-8_*_
#!/usr/bin/env python


import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('127.0.0.1',8080))

while True:
    msg=input('>>: ').strip()
    if not msg:continue

    s.send(msg.encode('utf-8'))
    data=s.recv(1024)
    print(data)

client

Thread類的其餘方法

Thread實例對象的方法
  # isAlive(): 返回線程是否活動的。
  # getName(): 返回線程名。
  # setName(): 設置線程名。

threading模塊提供的一些方法：
  # threading.currentThread(): 返回當前的線程變量。
  # threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啓動後、結束前，不包括啓動前和終止後的線程。
  # threading.activeCount(): 返回正在運行的線程數量，與len(threading.enumerate())有相同的結果。

from threading import Thread
import threading
from multiprocessing import Process
import os

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == '__main__':
    #在主進程下開啓線程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread()) #主線程
    print(threading.enumerate()) #連同主線程在內有兩個運行的線程
    print(threading.active_count())
    print('主線程/主進程')

    '''
    打印結果:
    MainThread
    <_MainThread(MainThread, started 140735268892672)>
    [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
    主線程/主進程
    Thread-1
    '''

代碼示例

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    t.join()
    print('主線程')
    print(t.is_alive())
    '''
    egon say hello
    主線程
    False
    '''

join方法

守護線程

不管是進程仍是線程，都遵循：守護xx會等待主xx運行完畢後被銷燬。須要強調的是：運行完畢並不是終止運行

#1.對主進程來講，運行完畢指的是主進程代碼運行完畢
#2.對主線程來講，運行完畢指的是主線程所在的進程內全部非守護線程通通運行完畢，主線程纔算運行完畢

#1 主進程在其代碼結束後就已經算運行完畢了（守護進程在此時就被回收）,而後主進程會一直等非守護的子進程都運行完畢後回收子進程的資源(不然會產生殭屍進程)，纔會結束，
#2 主線程在其餘非守護線程運行完畢後纔算運行完畢（守護線程在此時就被回收）。由於主線程的結束意味着進程的結束，進程總體的資源都將被回收，而進程必須保證非守護線程都運行完畢後才能結束。

詳細解釋

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.setDaemon(True) #必須在t.start()以前設置
    t.start()

    print('主線程')
    print(t.is_alive())
    '''
    主線程
    True
    '''

守護線程例1

from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


t1=Thread(target=foo)
t2=Thread(target=bar)

t1.daemon=True
t1.start()
t2.start()
print("main-------")

守護線程例2

回到頂部

鎖

鎖與GIL

同步鎖

from threading import Thread
import os,time
def work():
    global n
    temp=n
    time.sleep(0.1)
    n=temp-1
if __name__ == '__main__':
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #結果可能爲99

多個線程搶佔資源的狀況

import threading
R=threading.Lock()
R.acquire()
'''
對公共數據的操做
'''
R.release()

from threading import Thread,Lock
import os,time
def work():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #結果確定爲0，由原來的併發執行變成串行，犧牲了執行效率保證了數據安全

同步鎖的引用

#不加鎖:併發執行,速度快,數據不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print('%s is running' %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
'''


#不加鎖:未加鎖部分併發執行,加鎖部分串行執行,速度慢,數據安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    #未加鎖的代碼併發運行
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    #加鎖的代碼串行運行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
'''

#有的同窗可能有疑問:既然加鎖會讓運行變成串行,那麼我在start以後當即使用join,就不用加鎖了啊,也是串行的效果啊
#沒錯:在start以後馬上使用jion,確定會將100個任務的執行變成串行,毫無疑問,最終n的結果也確定是0,是安全的,但問題是
#start後當即join:任務內的全部代碼都是串行執行的,而加鎖,只是加鎖的部分即修改共享數據的部分是串行的
#單從保證數據安全方面,兩者均可以實現,但很明顯是加鎖的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == '__main__':
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗時是多麼的恐怖
'''

）

互斥鎖與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的次數，從而使得資源能夠被屢次require。直到一個線程全部的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

典型問題：科學家吃麪

import time
from threading import Thread,Lock
noodle_lock = Lock()
fork_lock = Lock()
def eat1(name):
    noodle_lock.acquire()
    print('%s 搶到了麪條'%name)
    fork_lock.acquire()
    print('%s 搶到了叉子'%name)
    print('%s 吃麪'%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print('%s 搶到了叉子' % name)
    time.sleep(1)
    noodle_lock.acquire()
    print('%s 搶到了麪條' % name)
    print('%s 吃麪' % name)
    noodle_lock.release()
    fork_lock.release()

for name in ['哪吒','egon','yuan']:
    t1 = Thread(target=eat1,args=(name,))
    t2 = Thread(target=eat2,args=(name,))
    t1.start()
    t2.start()

死鎖問題

import time
from threading import Thread,RLock
fork_lock = noodle_lock = RLock()
def eat1(name):
    noodle_lock.acquire()
    print('%s 搶到了麪條'%name)
    fork_lock.acquire()
    print('%s 搶到了叉子'%name)
    print('%s 吃麪'%name)
    fork_lock.release()
    noodle_lock.release()

def eat2(name):
    fork_lock.acquire()
    print('%s 搶到了叉子' % name)
    time.sleep(1)
    noodle_lock.acquire()
    print('%s 搶到了麪條' % name)
    print('%s 吃麪' % name)
    noodle_lock.release()
    fork_lock.release()

for name in ['哪吒','egon','yuan']:
    t1 = Thread(target=eat1,args=(name,))
    t2 = Thread(target=eat2,args=(name,))
    t1.start()
    t2.start()

遞歸鎖解決死鎖問題

回到頂部

線程隊列

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 queue

q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
結果(先進先出):
first
second
third
'''

先進先出

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

import queue

q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())
'''
結果(後進先出):
third
second
first
'''

後進先出

class queue.PriorityQueue(maxsize=0) #存儲數據時可設置優先級的隊列

import queue

q=queue.PriorityQueue()
#put進入一個元組,元組的第一個元素是優先級(一般是數字,也能夠是非數字之間的比較),數字越小優先級越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))

print(q.get())
print(q.get())
print(q.get())
'''
結果(數字越小優先級越高,優先級高的優先出隊):
(10, 'b')
(20, 'a')
(30, 'c')
'''

優先級隊列

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 ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ProcessPoolExecutor(max_workers=3)

    futures=[]
    for i in range(11):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)
    print('+++>')
    for future in futures:
        print(future.result())

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相同

ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit

map的用法

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def 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(3)
    # for url in urls:
    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)
    # p.close()
    # p.join()

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一個future對象obj，須要用obj.result()拿到結果