python-學習-python併發編程之多進程與多線程

時間 2019-11-17

原文原文鏈接

一 multiprocessing模塊介紹

python中的多線程沒法利用多核優點，若是想要充分地使用多核CPU的資源（os.cpu_count()查看），在python中大部分狀況須要使用多進程。Python提供了multiprocessing。
multiprocessing模塊用來開啓子進程，並在子進程中執行咱們定製的任務（好比函數），該模塊與多線程模塊threading的編程接口相似。html

　 multiprocessing模塊的功能衆多：支持子進程、通訊和共享數據、執行不一樣形式的同步，提供了Process、Queue、Pipe、Lock等組件。python

須要再次強調的一點是：與線程不一樣，進程沒有任何共享狀態，進程修改的數據，改動僅限於該進程內。數據庫

二 Process類的介紹

建立進程的類：編程

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化獲得的對象，表示一個子進程中的任務（還沒有啓動）

強調：
1. 須要使用關鍵字的方式來指定參數
2. args指定的爲傳給target函數的位置參數，是一個元組形式，必須有逗號

參數介紹：windows

1 group參數未使用，值始終爲None
2 
3 target表示調用對象，即子進程要執行的任務
4 
5 args表示調用對象的位置參數元組，args=(1,2,'egon',)
6 
7 kwargs表示調用對象的字典,kwargs={'name':'egon','age':18}
8 
9 name爲子進程的名稱

　 方法介紹：數組

 1 p.start()：啓動進程，並調用該子進程中的p.run() 
 2 p.run():進程啓動時運行的方法，正是它去調用target指定的函數，咱們自定義類的類中必定要實現該方法  
 3 
 4 p.terminate():強制終止進程p，不會進行任何清理操做，若是p建立了子進程，該子進程就成了殭屍進程，使用該方法須要特別當心這種狀況。若是p還保存了一個鎖那麼也將不會被釋放，進而致使死鎖
 5 p.is_alive():若是p仍然運行，返回True
 6 
 7 p.join([timeout]):主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，須要強調的是，p.join只能join住start開啓的進程，而不能join住run開啓的進程

屬性介紹：安全

1 p.daemon：默認值爲False，若是設爲True，表明p爲後臺運行的守護進程，當p的父進程終止時，p也隨之終止，而且設定爲True後，p不能建立本身的新進程，必須在p.start()以前設置
2 
3 p.name:進程的名稱
4 
5 p.pid：進程的pid
6 
7 p.exitcode:進程在運行時爲None、若是爲–N，表示被信號N結束(瞭解便可)
8 
9 p.authkey:進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是爲涉及網絡鏈接的底層進程間通訊提供安全性，這類鏈接只有在具備相同的身份驗證鍵時才能成功（瞭解便可）

三 Process類的使用

注意：在windows中Process()必須放到# if __name__ == '__main__':下服務器

詳細解釋

建立並開啓子進程的兩種方式網絡

方法一

方法二

練習1：把上週所學的socket通訊變成併發的形式多線程

server端

多個client端

這麼實現有沒有問題？？？

Process對象的join方法

join：主進程等，等待子進程結束

有了join，程序不就是串行了嗎？？？

Process對象的其餘方法或屬性（瞭解）

terminate與is_alive

name與pid

四守護進程

主進程建立守護進程

　　其一：守護進程會在主進程代碼執行結束後就終止

　　其二：守護進程內沒法再開啓子進程,不然拋出異常：AssertionError: daemonic processes are not allowed to have children

注意：進程之間是互相獨立的，主進程代碼運行結束，守護進程隨即終止

View Code

迷惑人的例子

五進程同步(鎖)

進程之間數據不共享,可是共享同一套文件系統,因此訪問同一個文件,或同一個打印終端,是沒有問題的,

競爭帶來的結果就是錯亂，如何控制，就是加鎖處理

part1：多個進程共享同一打印終端

併發運行,效率高,但競爭同一打印終端,帶來了打印錯亂

加鎖：由併發變成了串行,犧牲了運行效率,但避免了競爭

part2：多個進程共享同一文件

文件當數據庫，模擬搶票

併發運行，效率高，但競爭寫同一文件，數據寫入錯亂

加鎖：購票行爲由併發變成了串行，犧牲了運行效率，但保證了數據安全

總結：

加鎖能夠保證多個進程修改同一塊數據時，同一時間只能有一個任務能夠進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。
雖然能夠用文件共享數據實現進程間通訊，但問題是：
1.效率低
2.須要本身加鎖處理

爲此mutiprocessing模塊爲咱們提供了基於消息的IPC通訊機制：隊列和管道。
1 隊列和管道都是將數據存放於內存中
2 隊列又是基於（管道+鎖）實現的，可讓咱們從複雜的鎖問題中解脫出來，
咱們應該儘可能避免使用共享數據，儘量使用消息傳遞和隊列，避免處理複雜的同步和鎖問題，並且在進程數目增多時，每每能夠得到更好的可獲展性。

六隊列（推薦使用）

進程彼此之間互相隔離，要實現進程間通訊（IPC），multiprocessing模塊支持兩種形式：隊列和管道，這兩種方式都是使用消息傳遞的

　建立隊列的類（底層就是以管道和鎖定的方式實現）：

1 Queue([maxsize]):建立共享的進程隊列，Queue是多進程安全的隊列，可使用Queue實現多進程之間的數據傳遞。

參數介紹：

1 maxsize是隊列中容許最大項數，省略則無大小限制。

　 方法介紹：

主要方法：

1 q.put方法用以插入數據到隊列中，put方法還有兩個可選參數：blocked和timeout。若是blocked爲True（默認值），而且timeout爲正值，該方法會阻塞timeout指定的時間，直到該隊列有剩餘的空間。若是超時，會拋出Queue.Full異常。若是blocked爲False，但該Queue已滿，會當即拋出Queue.Full異常。
2 q.get方法能夠從隊列讀取而且刪除一個元素。一樣，get方法有兩個可選參數：blocked和timeout。若是blocked爲True（默認值），而且timeout爲正值，那麼在等待時間內沒有取到任何元素，會拋出Queue.Empty異常。若是blocked爲False，有兩種狀況存在，若是Queue有一個值可用，則當即返回該值，不然，若是隊列爲空，則當即拋出Queue.Empty異常.
3  
4 q.get_nowait():同q.get(False)
5 q.put_nowait():同q.put(False)
6 
7 q.empty():調用此方法時q爲空則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中又加入了項目。
8 q.full()：調用此方法時q已滿則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中的項目被取走。
9 q.qsize():返回隊列中目前項目的正確數量，結果也不可靠，理由同q.empty()和q.full()同樣

其餘方法(瞭解)：

1 q.cancel_join_thread():不會在進程退出時自動鏈接後臺線程。能夠防止join_thread()方法阻塞
2 q.close():關閉隊列，防止隊列中加入更多數據。調用此方法，後臺線程將繼續寫入那些已經入隊列但還沒有寫入的數據，但將在此方法完成時立刻關閉。若是q被垃圾收集，將調用此方法。關閉隊列不會在隊列使用者中產生任何類型的數據結束信號或異常。例如，若是某個使用者正在被阻塞在get()操做上，關閉生產者中的隊列不會致使get()方法返回錯誤。
3 q.join_thread()：鏈接隊列的後臺線程。此方法用於在調用q.close()方法以後，等待全部隊列項被消耗。默認狀況下，此方法由不是q的原始建立者的全部進程調用。調用q.cancel_join_thread方法能夠禁止這種行爲

　應用：

View Code

生產者消費者模型

在併發編程中使用生產者和消費者模式可以解決絕大多數併發問題。該模式經過平衡生產線程和消費線程的工做能力來提升程序的總體處理數據的速度。

爲何要使用生產者和消費者模式

在線程世界裏，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發當中，若是生產者處理速度很快，而消費者處理速度很慢，那麼生產者就必須等待消費者處理完，才能繼續生產數據。一樣的道理，若是消費者的處理能力大於生產者，那麼消費者就必須等待生產者。爲了解決這個問題因而引入了生產者和消費者模式。

什麼是生產者消費者模式

生產者消費者模式是經過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通信，而經過阻塞隊列來進行通信，因此生產者生產完數據以後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列裏取，阻塞隊列就至關於一個緩衝區，平衡了生產者和消費者的處理能力。

基於隊列實現生產者消費者模型

from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(q):
    for i in range(10):
        time.sleep(random.randint(1,3))
        res='包子%s' %i
        q.put(res)
        print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))

if __name__ == '__main__':
    q=Queue()
    #生產者們:即廚師們
    p1=Process(target=producer,args=(q,))

    #消費者們:即吃貨們
    c1=Process(target=consumer,args=(q,))

    #開始
    p1.start()
    c1.start()
    print('主')

此時的問題是主進程永遠不會結束，緣由是：生產者p在生產完後就結束了，可是消費者c在取空了q以後，則一直處於死循環中且卡在q.get()這一步。

解決方式無非是讓生產者在生產完畢後，往隊列中再發一個結束信號，這樣消費者在接收到結束信號後就能夠break出死循環

生產者在生產完畢後發送結束信號None

注意：結束信號None，不必定要由生產者發，主進程裏一樣能夠發，但主進程須要等生產者結束後才應該發送該信號

主進程在生產者生產完畢後發送結束信號None

但上述解決方式，在有多個生產者和多個消費者時，咱們則須要用一個很low的方式去解決

有幾個生產者就須要發送幾回結束信號：至關low

其實咱們的思路無非是發送結束信號而已，有另一種隊列提供了這種機制

   #JoinableQueue([maxsize])：這就像是一個Queue對象，但隊列容許項目的使用者通知生成者項目已經被成功處理。通知進程是使用共享的信號和條件變量來實現的。

   #參數介紹：
    maxsize是隊列中容許最大項數，省略則無大小限制。    
　 #方法介紹：
    JoinableQueue的實例p除了與Queue對象相同的方法以外還具備：
    q.task_done()：使用者使用此方法發出信號，表示q.get()的返回項目已經被處理。若是調用此方法的次數大於從隊列中刪除項目的數量，將引起ValueError異常
    q.join():生產者調用此方法進行阻塞，直到隊列中全部的項目均被處理。阻塞將持續到隊列中的每一個項目均調用q.task_done（）方法爲止

View Code

七管道

進程間通訊（IPC）方式二：管道（不推薦使用，瞭解便可）

介紹

基於管道實現進程間通訊（與隊列的方式是相似的，隊列就是管道加鎖實現的）

注意：生產者和消費者都沒有使用管道的某個端點，就應該將其關閉，如在生產者中關閉管道的右端，在消費者中關閉管道的左端。若是忘記執行這些步驟，程序可能再消費者中的recv()操做上掛起。管道是由操做系統進行引用計數的,必須在全部進程中關閉管道後才能生產EOFError異常。所以在生產者中關閉管道不會有任何效果，付費消費者中也關閉了相同的管道端點。

管道能夠用於雙向通訊，利用一般在客戶端/服務器中使用的請求／響應模型或遠程過程調用，就可使用管道編寫與進程交互的程序

八共享數據

展望將來，基於消息傳遞的併發編程是大勢所趨

即使是使用線程，推薦作法也是將程序設計爲大量獨立的線程集合

經過消息隊列交換數據。這樣極大地減小了對使用鎖定和其餘同步手段的需求，

還能夠擴展到分佈式系統中

進程間通訊應該儘可能避免使用本節所講的共享數據的方式

進程間數據是獨立的，能夠藉助於隊列或管道實現通訊，兩者都是基於消息傳遞的

雖然進程間數據獨立，但能夠經過Manager實現數據共享，事實上Manager的功能遠不止於此

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example,

進程之間操做共享的數據

九信號量(瞭解）

信號量Semahpore（同線程同樣）

十事件(瞭解)

Event（同線程同樣）

十一進程池

在利用Python進行系統管理的時候，特別是同時操做多個文件目錄，或者遠程控制多臺主機，並行操做能夠節約大量的時間。多進程是實現併發的手段之一，須要注意的問題是：

很明顯須要併發執行的任務一般要遠大於核數
一個操做系統不可能無限開啓進程，一般有幾個核就開幾個進程
進程開啓過多，效率反而會降低（開啓進程是須要佔用系統資源的，並且開啓多餘核數目的進程也沒法作到並行）

例如當被操做對象數目不大時，能夠直接利用multiprocessing中的Process動態成生多個進程，十幾個還好，但若是是上百個，上千個。。。手動的去限制進程數量卻又太過繁瑣，此時能夠發揮進程池的功效。

咱們就能夠經過維護一個進程池來控制進程數目，好比httpd的進程模式，規定最小進程數和最大進程數...
ps：對於遠程過程調用的高級應用程序而言，應該使用進程池，Pool能夠提供指定數量的進程，供用戶調用，當有新的請求提交到pool中時，若是池尚未滿，那麼就會建立一個新的進程用來執行該請求；但若是池中的進程數已經達到規定最大值，那麼該請求就會等待，直到池中有進程結束，就重用進程池中的進程。

建立進程池的類：若是指定numprocess爲3，則進程池會從無到有建立三個進程，而後自始至終使用這三個進程去執行全部任務，不會開啓其餘進程

1 Pool([numprocess  [,initializer [, initargs]]]):建立進程池

參數介紹：

1 numprocess:要建立的進程數，若是省略，將默認使用cpu_count()的值
2 initializer：是每一個工做進程啓動時要執行的可調用對象，默認爲None
3 initargs：是要傳給initializer的參數組

　 方法介紹：

主要方法：

1 p.apply(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。須要強調的是：此操做並不會在全部池工做進程中並執行func函數。若是要經過不一樣參數併發地執行func函數，必須從不一樣線程調用p.apply()函數或者使用p.apply_async()
2 p.apply_async(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變爲可用時，將理解傳遞給callback。callback禁止執行任何阻塞操做，不然將接收其餘異步操做中的結果。
3    
4 p.close():關閉進程池，防止進一步操做。若是全部操做持續掛起，它們將在工做進程終止前完成
5 P.jion():等待全部工做進程退出。此方法只能在close（）或teminate()以後調用

其餘方法（瞭解部分）

View Code

應用：

apply同步執行：阻塞式

apply_async異步執行：非阻塞

詳解：apply_async與apply

練習2：使用進程池維護固定數目的進程（重寫練習1）

server端

客戶端

發現：併發開啓多個客戶端，服務端同一時間只有3個不一樣的pid，幹掉一個客戶端，另一個客戶端纔會進來，被3個進程之一處理

　　回掉函數：

須要回調函數的場景：進程池中任何一個任務一旦處理完了，就當即告知主進程：我好了額，你能夠處理個人結果了。主進程則調用一個函數去處理該結果，該函數即回調函數

咱們能夠把耗時間（阻塞）的任務放到進程池中，而後指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。

View Code

爬蟲案例

　　若是在主進程中等待進程池中全部任務都執行完畢後，再統一處理結果，則無需回調函數

View Code

python併發編程之多線程理論部分：http://www.cnblogs.com/linhaifeng/articles/7430082.html

1、開啓線程的兩種方式

方式一

方式二

2、在一個進程下開啓多個線程與在一個進程下開啓多個子進程的區別

1 誰的開啓速度快

2 瞅一瞅pid

3 同一進程內的線程共享該進程的數據？

　　　　練習一：

多線程併發的socket服務端

客戶端

　　　　練習二：三個任務，一個接收用戶輸入，一個將用戶輸入的內容格式化成大寫，一個將格式化後的結果存入文件

View Code

3、守護線程

不管是進程仍是線程，都遵循：守護xxx會等待主xxx運行完畢後被銷燬

須要強調的是：運行完畢並不是終止運行

#1.對主進程來講，運行完畢指的是主進程代碼運行完畢

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

詳細解釋：

#1 主進程在其代碼結束後就已經算運行完畢了（守護進程在此時就被回收）,而後主進程會一直等非守護的子進程都運行完畢後回收子進程的資源(不然會產生殭屍進程)，纔會結束，

#2 主線程在其餘非守護線程運行完畢後纔算運行完畢（守護線程在此時就被回收）。由於主線程的結束意味着進程的結束，進程總體的資源都將被回收，而進程必須保證非守護線程都運行完畢後才能結束。

 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()以前設置
10     t.start()
11 
12     print('主線程')
13     print(t.is_alive())
14     '''
15     主線程
16     True
17     '''

例子

迷惑人的例子

4、同步鎖

三個須要注意的點：
#1.線程搶的是GIL鎖，GIL鎖至關於執行權限，拿到執行權限後才能拿到互斥鎖Lock，其餘線程也能夠搶到GIL，但若是發現Lock仍然沒有被釋放則阻塞，即使是拿到執行權限GIL也要馬上交出來

#2.join是等待全部，即總體串行，而鎖只是鎖住修改共享數據的部分，即部分串行，要想保證數據安全的根本原理在於讓併發變成串行，join與互斥鎖均可以實現，毫無疑問，互斥鎖的部分串行效率要更高

#3. 必定要看本小節最後的GIL與互斥鎖的經典分析    GIL：連接：http://www.cnblogs.com/linhaifeng/articles/7449853.html

GIL VS Lock

Python已經有一個GIL來保證同一時間只能有一個線程來執行了，爲何這裏還須要lock?

　首先咱們須要達成共識：鎖的目的是爲了保護共享的數據，同一時間只能有一個線程來修改共享的數據

而後，咱們能夠得出結論：保護不一樣的數據就應該加不一樣的鎖。

　最後，問題就很明朗了，GIL 與Lock是兩把鎖，保護的數據不同，前者是解釋器級別的（固然保護的就是解釋器級別的數據，好比垃圾回收的數據），後者是保護用戶本身開發的應用程序的數據，很明顯GIL不負責這件事，只能用戶自定義加鎖處理，即Lock

過程分析：全部線程搶的是GIL鎖，或者說全部線程搶的是執行權限

　　線程1搶到GIL鎖，拿到執行權限，開始執行，而後加了一把Lock，尚未執行完畢，即線程1還未釋放Lock，有可能線程2搶到GIL鎖，開始執行，執行過程當中發現Lock尚未被線程1釋放，因而線程2進入阻塞，被奪走執行權限，有可能線程1拿到GIL，而後正常執行到釋放Lock。。。這就致使了串行運行的效果

　　既然是串行，那咱們執行

　　t1.start()

　　t1.join

　　t2.start()

　　t2.join()

　　這也是串行執行啊，爲什麼還要加Lock呢，需知join是等待t1全部的代碼執行完，至關於鎖住了t1的全部代碼，而Lock只是鎖住一部分操做共享數據的代碼。

詳細

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

鎖一般被用來實現對共享資源的同步訪問。爲每個共享資源建立一個Lock對象，當你須要訪問該資源時，調用acquire方法來獲取鎖對象（若是其它線程已經得到了該鎖，則當前線程需等待其被釋放），待資源訪問完後，再調用release方法釋放鎖：

import threading

R=threading.Lock()

R.acquire()
'''
對公共數據的操做
'''
R.release()

View Code

GIL鎖與互斥鎖綜合分析（重點！！！）

互斥鎖與join的區別（重點！！！）

　　1.死鎖現象與遞歸鎖

進程也有死鎖與遞歸鎖，在進程那裏忘記說了，放到這裏一切說了額

所謂死鎖：是指兩個或兩個以上的進程或線程在執行過程當中，因爭奪資源而形成的一種互相等待的現象，若無外力做用，它們都將沒法推動下去。此時稱系統處於死鎖狀態或系統產生了死鎖，這些永遠在互相等待的進程稱爲死鎖進程，以下就是死鎖

View Code

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

這個RLock內部維護着一個Lock和一個counter變量，counter記錄了acquire的次數，從而使得資源能夠被屢次require。直到一個線程全部的acquire都被release，其餘的線程才能得到資源。上面的例子若是使用RLock代替Lock，則不會發生死鎖：

mutexA=mutexB=threading.RLock() #一個線程拿到鎖，counter加1,該線程內又碰到加鎖的狀況，則counter繼續加1，這期間全部其餘線程都只能等待，等待該線程釋放全部鎖，即counter遞減到0爲止

5、定時器

定時器，指定n秒後執行某操做

from threading import Timer
 
 
def hello():
    print("hello, world")
 
t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

6、paramiko模塊

1. 介紹：

paramiko是一個用於作遠程控制的模塊，使用該模塊能夠對遠程服務器進行命令或文件操做，值得一說的是，fabric和ansible內部的遠程管理就是使用的paramiko來現實。

2. 下載安裝

pip3 install paramiko #在python3中

在python2中

3. 使用

SSHClient

用於鏈接遠程服務器並執行基本命令

基於用戶名密碼鏈接：

import paramiko

# 建立SSH對象
ssh = paramiko.SSHClient()
# 容許鏈接不在know_hosts文件中的主機
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 鏈接服務器
ssh.connect(hostname='120.92.84.249', port=22, username='root', password='xxx')

# 執行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 獲取命令結果
result = stdout.read()
print(result.decode('utf-8'))
# 關閉鏈接
ssh.close()

SSHClient 封裝 Transport

基於公鑰密鑰鏈接：

客戶端文件名：id_rsa

服務端必須有文件名：authorized_keys(在用ssh-keygen時，必須製做一個authorized_keys,能夠用ssh-copy-id來製做)

View Code

SSHClient 封裝 Transport

基於私鑰字符串進行鏈接

SFTPClient

用於鏈接遠程服務器並執行上傳下載

基於用戶名密碼上傳下載

View Code

基於公鑰密鑰上傳下載