python並發編程&多進程(二)

本文轉載自查看原文 2017-12-05 18:54 3898 python並發編程

一 multiprocessing模塊介紹

python中的多線程無法利用多核優勢，如果想要充分地使用多核CPU的資源（os.cpu_count()查看本機核數），在python中大部分情況需要使用多進程。

　Python提供了multiprocessing。
multiprocessing模塊用來開啟子進程，並在子進程中執行我們定制的任務（比如函數），該模塊與多線程模塊threading的編程接口類似。

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

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

二 Process類的介紹

創建進程的類：

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化得到的對象，表示一個子進程中的任務（尚未啟動）

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

參數介紹：

group參數未使用，值始終為None

target表示調用對象，即子進程要執行的任務

args表示調用對象的位置參數元組，args=(1,2,'egon',)

kwargs表示調用對象的字典,kwargs={'name':'egon','age':18}

name為子進程的名稱

　 方法介紹：

p.start()：啟動進程，並調用該子進程中的p.run() 
p.run():進程啟動時運行的方法，正是它去調用target指定的函數，我們自定義類的類中一定要實現該方法  

p.terminate():強制終止進程p，不會進行任何清理操作，如果p創建了子進程，該子進程就成了僵屍進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那么也將不會被釋放，進而導致死鎖
p.is_alive():如果p仍然運行，返回True
 
p.join([timeout]):主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，需要強調的是，p.join只能join住start開啟的進程，而不能join住run開啟的進程

屬性介紹：

p.daemon：默認值為False，如果設為True，代表p為后台運行的守護進程，當p的父進程終止時，p也隨之終止，並且設定為True后，p不能創建自己的新進程，必須在p.start()之前設置

p.name:進程的名稱

p.pid：進程的pid

p.exitcode:進程在運行時為None、如果為–N，表示被信號N結束(了解即可)

p.authkey:進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是為涉及網絡連接的底層進程間通信提供安全性，這類連接只有在具有相同的身份驗證鍵時才能成功（了解即可）

三 Process類的使用

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

Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module. 
If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources). 
This is the reason for hiding calls to Process() inside

if __name__ == "__main__"
since statements inside this if-statement will not get called upon import.
由於Windows沒有fork，多處理模塊啟動一個新的Python進程並導入調用模塊。 
如果在導入時調用Process（），那么這將啟動無限繼承的新進程（或直到機器耗盡資源）。 
這是隱藏對Process（）內部調用的原，使用if __name__ == “__main __”，這個if語句中的語句將不會在導入時被調用。

創建並開啟子進程的兩種方式

import time
import random
from multiprocessing import Process
def piao(name):
    print('%s piaoing' %name)
    time.sleep(random.randrange(1,5))
    print('%s piao end' %name)



p1=Process(target=piao,args=('egon',)) #必須加,號
p2=Process(target=piao,args=('alex',))
p3=Process(target=piao,args=('wupeqi',))
p4=Process(target=piao,args=('yuanhao',))

p1.start()
p2.start()
p3.start()
p4.start()
print('主線程')

方法一

import time
import random
from multiprocessing import Process


class Piao(Process):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('%s piaoing' %self.name)

        time.sleep(random.randrange(1,5))
        print('%s piao end' %self.name)

p1=Piao('egon')
p2=Piao('alex')
p3=Piao('wupeiqi')
p4=Piao('yuanhao')

p1.start() #start會自動調用run
p2.start()
p3.start()
p4.start()
print('主線程')

方法二

進程直接的內存空間是隔離的

from multiprocessing import Process
n=100 #在windows系統中應該把全局變量定義在if __name__ == '__main__'之上就可以了
def work():
    global n
    n=0
    print('子進程內: ',n)


if __name__ == '__main__':
    p=Process(target=work)
    p.start()
    print('主進程內: ',n)

空間隔離示例

練習1：把socket通信變成並發的形式

from socket import *
from multiprocessing import Process

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn,client_addr):
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__': #windows下start進程一定要寫到這下面
    while True:
        conn,client_addr=server.accept()
        p=Process(target=talk,args=(conn,client_addr))
        p.start()

server端

from socket import *

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


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

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

多個client端

每來一個客戶端，都在服務端開啟一個進程，如果並發來一個萬個客戶端，要開啟一萬個進程嗎，你自己嘗試着在你自己的機器上開啟一萬個，10萬個進程試一試。
解決方法：進程池

存在的問題

Process對象的join方法

from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()
    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,3))
        print('%s is piao end' %self.name)


p=Piao('egon')
p.start()
p.join(0.0001) #等待p停止,等0.0001秒就不再等了
print('開始')

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

from multiprocessing import Process
import time
import random
def piao(name):
    print('%s is piaoing' %name)
    time.sleep(random.randint(1,3))
    print('%s is piao end' %name)

p1=Process(target=piao,args=('egon',))
p2=Process(target=piao,args=('alex',))
p3=Process(target=piao,args=('yuanhao',))
p4=Process(target=piao,args=('wupeiqi',))

p1.start()
p2.start()
p3.start()
p4.start()

#有的同學會有疑問:既然join是等待進程結束,那么我像下面這樣寫,進程不就又變成串行的了嗎?
#當然不是了,必須明確：p.join()是讓誰等？
#很明顯p.join()是讓主線程等待p的結束，卡住的是主線程而絕非進程p，

#詳細解析如下：
#進程只要start就會在開始運行了,所以p1-p4.start()時,系統中已經有四個並發的進程了
#而我們p1.join()是在等p1結束,沒錯p1只要不結束主線程就會一直卡在原地,這也是問題的關鍵
#join是讓主線程等,而p1-p4仍然是並發執行的,p1.join的時候,其余p2,p3,p4仍然在運行,等#p1.join結束,可能p2,p3,p4早已經結束了,這樣p2.join,p3.join.p4.join直接通過檢測，無需等待
# 所以4個join花費的總時間仍然是耗費時間最長的那個進程運行的時間
p1.join()
p2.join()
p3.join()
p4.join()

print('主線程')


#上述啟動進程與join進程可以簡寫為
# p_l=[p1,p2,p3,p4]
# 
# for p in p_l:
#     p.start()
# 
# for p in p_l:
#     p.join()

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

Process對象的其他方法或屬性（了解）

#進程對象的其他方法一:terminate,is_alive
from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s is piao end' %self.name)


p1=Piao('egon1')
p1.start()

p1.terminate()#關閉進程,不會立即關閉,所以is_alive立刻查看的結果可能還是存活
print(p1.is_alive()) #結果為True

print('開始')
print(p1.is_alive()) #結果為False

terminate與is_alive

from multiprocessing import Process
import time
import random
class Piao(Process):
    def __init__(self,name):
        # self.name=name
        # super().__init__() #Process的__init__方法會執行self.name=Piao-1,
        #                    #所以加到這里,會覆蓋我們的self.name=name

        #為我們開啟的進程設置名字的做法
        super().__init__()
        self.name=name

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,3))
        print('%s is piao end' %self.name)

p=Piao('egon')
p.start()
print('開始')
print(p.pid) #查看pid

name與pid

四守護進程

主進程創建守護進程

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

　　其二：守護進程內無法再開啟子進程,否則拋出異常：AssertionError: daemonic processes are not allowed to have children

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

from multiprocessing import Process
import time
import random

class Piao(Process):
    def __init__(self,name):
        self.name=name
        super().__init__()
    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randrange(1,3))
        print('%s is piao end' %self.name)


p=Piao('egon')
p.daemon=True #一定要在p.start()前設置,設置p為守護進程,禁止p創建子進程,並且父進程代碼執行結束,p即終止運行
p.start()
print('主')

示例

#主進程代碼運行完畢,守護進程就會結束
from multiprocessing import Process
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

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


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
print("main-------") #打印該行則主進程代碼結束,則守護進程p1應該被終止,可能會有p1任務執行的打印信息123,因為主進程打印main----時,p1也執行了,但是隨即被終止

迷惑人的例子

五進程同步(鎖)

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

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

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

#並發運行,效率高,但競爭同一打印終端,帶來了打印錯亂
from multiprocessing import Process
import os,time
def work():
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())

if __name__ == '__main__':
    for i in range(3):
        p=Process(target=work)
        p.start()

並發運行，效率高，但競爭同一打印終端，致使打印錯亂

from multiprocessing import Process,Lock
import os,time
def work(lock):
    lock.acquire()
    print('%s is running' %os.getpid())
    time.sleep(2)
    print('%s is done' %os.getpid())
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,))
        p.start()

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

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

文件當數據庫，模擬搶票：

#文件db的內容為：{"count":1}
#注意一定要用雙引號，不然json無法識別
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩余票數%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模擬讀數據的網絡延遲
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模擬寫數據的網絡延遲
        json.dump(dic,open('db.txt','w'))
        print('\033[43m購票成功\033[0m')

def task(lock):
    search()
    get()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模擬並發100個客戶端搶票
        p=Process(target=task,args=(lock,))
        p.start()

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

#文件db的內容為：{"count":1}
#注意一定要用雙引號，不然json無法識別
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩余票數%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模擬讀數據的網絡延遲
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模擬寫數據的網絡延遲
        json.dump(dic,open('db.txt','w'))
        print('\033[43m購票成功\033[0m')

def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模擬並發100個客戶端搶票
        p=Process(target=task,args=(lock,))
        p.start()

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

總結：

#加鎖可以保證多個進程修改同一塊數據時，同一時間只能有一個任務可以進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。
雖然可以用文件共享數據實現進程間通信，但問題是：
1.效率低（共享數據基於文件，而文件是硬盤上的數據）
2.需要自己加鎖處理


#因此我們最好找尋一種解決方案能夠兼顧：1、效率高（多個進程共享一塊內存的數據）2、幫我們處理好鎖問題。這就是mutiprocessing模塊為我們提供的基於消息的IPC通信機制：隊列和管道。
1 隊列和管道都是將數據存放於內存中
2 隊列又是基於（管道+鎖）實現的，可以讓我們從復雜的鎖問題中解脫出來，
我們應該盡量避免使用共享數據，盡可能使用消息傳遞和隊列，避免處理復雜的同步和鎖問題，而且在進程數目增多時，往往可以獲得更好的可獲展性。

六隊列（推薦使用）

進程彼此之間互相隔離，要實現進程間通信（IPC——Inter-Process Communication，進程間通信）

　multiprocessing模塊支持兩種形式：隊列和管道，這兩種方式都是使用消息傳遞的

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

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

參數介紹：

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

方法介紹：

　　主要方法：

q.put方法用以插入數據到隊列中，put方法還有兩個可選參數：blocked和timeout。如果blocked為True（默認值），並且timeout為正值，該方法會阻塞timeout指定的時間，直到該隊列有剩余的空間。
　　　　　　如果超時，會拋出Queue.Full異常。如果blocked為False，但該Queue已滿，會立即拋出Queue.Full異常。
q.get方法可以從隊列讀取並且刪除一個元素。同樣，get方法有兩個可選參數：blocked和timeout。如果blocked為True（默認值），並且timeout為正值，那么在等待時間內沒有取到任何元素會拋出Queue.Empty異常。
　　　　　　如果blocked為False，有兩種情況存在，如果Queue有一個值可用，則立即返回該值，否則，如果隊列為空，則立即拋出Queue.Empty異常.

q.get_nowait():同q.get(False)
q.put_nowait():同q.put(False)

q.empty():調用此方法時q為空則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中又加入了項目。
q.full()：調用此方法時q已滿則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中的項目被取走。
q.qsize():返回隊列中目前項目的正確數量，結果也不可靠，理由同q.empty()和q.full()一樣

其他方法(了解)：

q.cancel_join_thread():不會在進程退出時自動連接后台線程。可以防止join_thread()方法阻塞
q.close():關閉隊列，防止隊列中加入更多數據。調用此方法，后台線程將繼續寫入那些已經入隊列但尚未寫入的數據，但將在此方法完成時馬上關閉。如果q被垃圾收集，將調用此方法。
　　　　　　關閉隊列不會在隊列使用者中產生任何類型的數據結束信號或異常。例如，如果某個使用者正在被阻塞在get()操作上，關閉生產者中的隊列不會導致get()方法返回錯誤。
q.join_thread()：連接隊列的后台線程。此方法用於在調用q.close()方法之后，等待所有隊列項被消耗。默認情況下，此方法由不是q的原始創建者的所有進程調用。
　　　　　　       調用q.cancel_join_thread方法可以禁止這種行為

　應用：

'''
multiprocessing模塊支持進程間通信的兩種主要形式:管道和隊列
都是基於消息傳遞實現的,但是隊列接口
'''

from multiprocessing import Process,Queue
import time
q=Queue(3)


#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
print(q.full()) #滿了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了

生產者消費者模型

在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡生產線程和消費線程的工作能力來提高程序的整體處理數據的速度。

為什么要使用生產者和消費者模式

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

什么是生產者消費者模式

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

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

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出死循環

from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        if res is None:break #收到結束信號則結束
        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))
    q.put(None) #發送結束信號
if __name__ == '__main__':
    q=Queue()
    #生產者們:即廚師們
    p1=Process(target=producer,args=(q,))

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

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

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

注意：結束信號None，不一定要由生產者發，主進程里同樣可以發，但主進程需要等生產者結束后才應該發送該信號

from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        if res is None:break #收到結束信號則結束
        time.sleep(random.randint(1,3))
        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(q):
    for i in range(2):
        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()

    p1.join()
    q.put(None) #發送結束信號
    print('主')

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

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

from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        if res is None:break #收到結束信號則結束
        time.sleep(random.randint(1,3))
        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

def producer(name,q):
    for i in range(2):
        time.sleep(random.randint(1,3))
        res='%s%s' %(name,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))
    p2=Process(target=producer,args=('骨頭',q))
    p3=Process(target=producer,args=('泔水',q))

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

    #開始
    p1.start()
    p2.start()
    p3.start()
    c1.start()

    p1.join() #必須保證生產者全部生產完畢,才應該發送結束信號
    p2.join()
    p3.join()
    q.put(None) #有幾個消費者就應該發送幾次結束信號None
    q.put(None) #發送結束信號
    print('主')

有幾個消費者就需要發送幾次結束信號：很low~

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

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

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

from multiprocessing import Process,JoinableQueue
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))

        q.task_done() #向q.join()發送一次信號,證明一個數據已經被取走了

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


if __name__ == '__main__':
    q=JoinableQueue()
    #生產者們:即廚師們
    p1=Process(target=producer,args=('包子',q))
    p2=Process(target=producer,args=('骨頭',q))
    p3=Process(target=producer,args=('泔水',q))

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

    #開始
    p_l=[p1,p2,p3,c1,c2]
    for p in p_l:
        p.start()

    p1.join()
    p2.join()
    p3.join()
    print('主') 
    
    #主進程等--->p1,p2,p3等---->c1,c2
    #p1,p2,p3結束了,證明c1,c2肯定全都收完了p1,p2,p3發到隊列的數據
    #因而c1,c2也沒有存在的價值了,應該隨着主進程的結束而結束,所以設置成守護進程

示例

七管道

進程間通信（IPC）方式二：管道（不推薦使用，了解即可）

#創建管道的類：
Pipe([duplex]):在進程之間創建一條管道，並返回元組（conn1,conn2）,其中conn1，conn2表示管道兩端的連接對象，強調一點：必須在產生Process對象之前產生管道
#參數介紹：
dumplex:默認管道是全雙工的，如果將duplex射成False，conn1只能用於接收，conn2只能用於發送。
#主要方法：
    conn1.recv():接收conn2.send(obj)發送的對象。如果沒有消息可接收，recv方法會一直阻塞。如果連接的另外一端已經關閉，那么recv方法會拋出EOFError。
    conn1.send(obj):通過連接發送對象。obj是與序列化兼容的任意對象
 #其他方法：
conn1.close():關閉連接。如果conn1被垃圾回收，將自動調用此方法
conn1.fileno():返回連接使用的整數文件描述符
conn1.poll([timeout]):如果連接上的數據可用，返回True。timeout指定等待的最長時限。如果省略此參數，方法將立即返回結果。如果將timeout射成None，操作將無限期地等待數據到達。
 
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法發送的一條完整的字節消息。maxlength指定要接收的最大字節數。如果進入的消息，超過了這個最大值，將引發IOError異常，並且在連接上無法進行進一步讀取。如果連接的另外一端已經關閉，再也不存在任何數據，將引發EOFError異常。
conn.send_bytes(buffer [, offset [, size]])：通過連接發送字節數據緩沖區，buffer是支持緩沖區接口的任意對象，offset是緩沖區中的字節偏移量，而size是要發送字節數。結果數據以單條消息的形式發出，然后調用c.recv_bytes()函數進行接收    
 
conn1.recv_bytes_into(buffer [, offset]):接收一條完整的字節消息，並把它保存在buffer對象中，該對象支持可寫入的緩沖區接口（即bytearray對象或類似的對象）。offset指定緩沖區中放置消息處的字節位移。返回值是收到的字節數。如果消息長度大於可用的緩沖區空間，將引發BufferTooShort異常。

介紹

from multiprocessing import Process,Pipe

import time,os
def consumer(p,name):
    left,right=p
    left.close()
    while True:
        try:
            baozi=right.recv()
            print('%s 收到包子:%s' %(name,baozi))
        except EOFError:
            right.close()
            break
def producer(seq,p):
    left,right=p
    right.close()
    for i in seq:
        left.send(i)
        # time.sleep(1)
    else:
        left.close()
if __name__ == '__main__':
    left,right=Pipe()

    c1=Process(target=consumer,args=((left,right),'c1'))
    c1.start()


    seq=(i for i in range(10))
    producer(seq,(left,right))

    right.close()
    left.close()

    c1.join()
    print('主進程')

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

注意：

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

from multiprocessing import Process,Pipe

import time,os
def adder(p,name):
    server,client=p
    client.close()
    while True:
        try:
            x,y=server.recv()
        except EOFError:
            server.close()
            break
        res=x+y
        server.send(res)
    print('server done')
if __name__ == '__main__':
    server,client=Pipe()

    c1=Process(target=adder,args=((server,client),'c1'))
    c1.start()

    server.close()

    client.send((10,20))
    print(client.recv())
    client.close()

    c1.join()
    print('主進程')
#注意：send()和recv()方法使用pickle模塊對對象進行序列化。

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

八共享數據

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

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

通過消息隊列交換數據。這樣極大地減少了對使用鎖定和其他同步手段的需求，還可以擴展到分布式系統中

進程間通信應該盡量避免使用本節所講的共享數據的方式

進程間數據是獨立的，可以借助於隊列或管道實現通信，二者都是基於消息傳遞的

雖然進程間數據獨立，但可以通過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.

from multiprocessing import Manager,Process,Lock
import os
def work(d,lock):
    # with lock: #不加鎖而操作共享的數據,肯定會出現數據錯亂
        d['count']-=1

if __name__ == '__main__':
    lock=Lock()
    with Manager() as m:
        dic=m.dict({'count':100})
        p_l=[]
        for i in range(100):
            p=Process(target=work,args=(dic,lock))
            p_l.append(p)
            p.start()
        for p in p_l:
            p.join()
        print(dic)
        #{'count': 94}

進程之間操作共享的數據

九信號量(了解）

互斥鎖 同時只允許一個線程更改數據，而Semaphore是同時允許一定數量的線程更改數據 ，比如廁所有3個坑，那最多只允許3個人上廁所，后面的人只能等里面有人出來了才能再進去，如果指定信號量為3，那么來一個人獲得一把鎖，計數加1，當計數等於3時，后面的人均需要等待。一旦釋放，就有人可以獲得一把鎖

    信號量與進程池的概念很像，但是要區分開，信號量涉及到加鎖的概念

from multiprocessing import Process,Semaphore
import time,random

def go_wc(sem,user):
    sem.acquire()
    print('%s 占到一個茅坑' %user)
    time.sleep(random.randint(0,3)) #模擬每個人拉屎速度不一樣，0代表有的人蹲下就起來了
    sem.release()

if __name__ == '__main__':
    sem=Semaphore(5)
    p_l=[]
    for i in range(13):
        p=Process(target=go_wc,args=(sem,'user%s' %i,))
        p.start()
        p_l.append(p)

    for i in p_l:
        i.join()
    print('============》')

信號量Semahpore（同線程一樣）

十事件(了解)

python線程的事件用於主線程控制其他線程的執行，事件主要提供了三個方法 set、wait、clear。

    事件處理的機制：全局定義了一個“Flag”，如果“Flag”值為 False，那么當程序執行 event.wait 方法時就會阻塞，如果“Flag”值為True，那么event.wait 方法時便不再阻塞。

clear：將“Flag”設置為False
set：將“Flag”設置為True
 

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

from multiprocessing import Process,Event
import time,random

def car(e,n):
    while True:
        if not e.is_set(): #Flase
            print('\033[31m紅燈亮\033[0m，car%s等着' %n)
            e.wait()
            print('\033[32m車%s 看見綠燈亮了\033[0m' %n)
            time.sleep(random.randint(3,6))
            if not e.is_set():
                continue
            print('走你,car', n)
            break

def police_car(e,n):
    while True:
        if not e.is_set():
            print('\033[31m紅燈亮\033[0m，car%s等着' % n)
            e.wait(1)
            print('燈的是%s，警車走了,car %s' %(e.is_set(),n))
            break

def traffic_lights(e,inverval):
    while True:
        time.sleep(inverval)
        if e.is_set():
            e.clear() #e.is_set() ---->False
        else:
            e.set()

if __name__ == '__main__':
    e=Event()
    # for i in range(10):
    #     p=Process(target=car,args=(e,i,))
    #     p.start()

    for i in range(5):
        p = Process(target=police_car, args=(e, i,))
        p.start()
    t=Process(target=traffic_lights,args=(e,10))
    t.start()

    print('============》')

Event（同線程一樣）

十一進程池

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

很明顯需要並發執行的任務通常要遠大於核數

一個操作系統不可能無限開啟進程，通常有幾個核就開幾個進程

進程開啟過多，效率反而會下降（開啟進程是需要占用系統資源的，而且開啟多余核數目的進程也無法做到並行）

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

我們就可以通過維護一個進程池來控制進程數目，比如httpd的進程模式，規定最小進程數和最大進程數...

ps：對於遠程過程調用的高級應用程序而言，應該使用進程池，Pool可以提供指定數量的進程，供用戶調用，當有新的請求提交到pool中時，如果池還沒有滿，那么就會創建一個新的進程用來執行該請求；但如果池中的進程數已經達到規定最大值，那么該請求就會等待，直到池中有進程結束，就重用進程池中的進程。

創建進程池的類：

如果指定numprocess為3，則進程池會從無到有創建三個進程，然后自始至終使用這三個進程去執行所有任務，不會開啟其他進程

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

參數介紹：

numprocess:要創建的進程數，如果省略，將默認使用cpu_count()的值
initializer：是每個工作進程啟動時要執行的可調用對象，默認為None
initargs：是要傳給initializer的參數組

　 方法介紹：

主要方法：

p.apply(func [, args [, kwargs]]):在一個池工作進程中執行func(*args,**kwargs),然后返回結果。
　　需要強調的是：此操作並不會在所有池工作進程中並執行func函數。如果要通過不同參數並發地執行func函數，必須從不同線程調用p.apply()函數或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]]):在一個池工作進程中執行func(*args,**kwargs),然后返回結果。
　　此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變為可用時，將理解傳遞給callback。callback禁止執行任何阻塞操作，否則將接收其他異步操作中的結果。
    
p.close():關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成
P.jion():等待所有工作進程退出。此方法只能在close（）或teminate()之后調用

其他方法（了解部分）

方法apply_async()和map_async（）的返回值是AsyncResul的實例obj。實例具有以下方法
obj.get():返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內還沒有到達，將引發一場。如果遠程操作中引發了異常，它將在調用此方法時再次被引發。
obj.ready():如果調用完成，返回True
obj.successful():如果調用完成且沒有引發異常，返回True，如果在結果就緒之前調用此方法，引發異常
obj.wait([timeout]):等待結果變為可用。
obj.terminate()：立即終止所有工作進程，同時不執行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數

應用：

from multiprocessing import Pool
import os,time
def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3) #進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務
    res_l=[]
    for i in range(10):
        res=p.apply(work,args=(i,)) #同步調用，直到本次任務執行完畢拿到res，等待任務work執行的過程中可能有阻塞也可能沒有阻塞，但不管該任務是否存在阻塞，同步調用都會在原地等着，只是等的過程中若是任務發生了阻塞就會被奪走cpu的執行權限
        res_l.append(res)
    print(res_l)

同步調用apply

from multiprocessing import Pool
import os,time
def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3) #進程池中從無到有創建三個進程,以后一直是這三個進程在執行任務
    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,)) #同步運行,阻塞、直到本次任務執行完畢拿到res
        res_l.append(res)

    #異步apply_async用法：如果使用異步提交的任務，主進程需要使用jion，等待進程池內任務都處理完，然后可以用get收集結果，否則，主進程結束，進程池可能還沒來得及執行，也就跟着一起結束了
    p.close()
    p.join()
    for res in res_l:
        print(res.get()) #使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執行,立刻獲取結果,也根本無需get

異步調用apply_async

#一：使用進程池（異步調用,apply_async）
#coding: utf-8
from multiprocessing import Process,Pool
import time

def func(msg):
    print( "msg:", msg)
    time.sleep(1)
    return msg

if __name__ == "__main__":
    pool = Pool(processes = 3)
    res_l=[]
    for i in range(10):
        msg = "hello %d" %(i)
        res=pool.apply_async(func, (msg, ))   #維持執行的進程總數為processes，當一個進程執行完畢后會添加新的進程進去
        res_l.append(res)
    print("==============================>") #沒有后面的join，或get，則程序整體結束，進程池中的任務還沒來得及全部執行完也都跟着主進程一起結束了

    pool.close() #關閉進程池，防止進一步操作。如果所有操作持續掛起，它們將在工作進程終止前完成
    pool.join()   #調用join之前，先調用close函數，否則會出錯。執行完close后不會有新的進程加入到pool,join函數等待所有子進程結束

    print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>對象組成的列表,而非最終的結果,但這一步是在join后執行的,證明結果已經計算完畢,剩下的事情就是調用每個對象下的get方法去獲取結果
    for i in res_l:
        print(i.get()) #使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執行,立刻獲取結果,也根本無需get

#二：使用進程池（同步調用,apply）
#coding: utf-8
from multiprocessing import Process,Pool
import time

def func(msg):
    print( "msg:", msg)
    time.sleep(0.1)
    return msg

if __name__ == "__main__":
    pool = Pool(processes = 3)
    res_l=[]
    for i in range(10):
        msg = "hello %d" %(i)
        res=pool.apply(func, (msg, ))   #維持執行的進程總數為processes，當一個進程執行完畢后會添加新的進程進去
        res_l.append(res) #同步執行，即執行完一個拿到結果，再去執行另外一個
    print("==============================>")
    pool.close()
    pool.join()   #調用join之前，先調用close函數，否則會出錯。執行完close后不會有新的進程加入到pool,join函數等待所有子進程結束

    print(res_l) #看到的就是最終的結果組成的列表
    for i in res_l: #apply是同步的，所以直接得到結果，沒有get()方法
        print(i)

詳解：apply_async與apply

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

#Pool內的進程數默認是cpu核數，假設為4（查看方法os.cpu_count()）
#開啟6個客戶端，會發現2個客戶端處於等待狀態
#在每個進程內查看pid，會發現pid使用為4個，即多個客戶端公用4個進程
from socket import *
from multiprocessing import Pool
import os

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn,client_addr):
    print('進程pid: %s' %os.getpid())
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__':
    p=Pool()
    while True:
        conn,client_addr=server.accept()
        p.apply_async(talk,args=(conn,client_addr))
        # p.apply(talk,args=(conn,client_addr)) #同步的話，則同一時間只有一個客戶端能訪問

server端

from socket import *

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


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

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

客戶端

發現：

　　並發開啟多個客戶端，服務端同一時間只有3個不同的pid，干掉一個客戶端，另外一個客戶端才會進來，被3個進程之一處理

回掉函數：

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

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

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 pasrse_page(res):
    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)
    res_l=[]
    for url in urls:
        res=p.apply_async(get_page,args=(url,),callback=pasrse_page)
        res_l.append(res)

    p.close()
    p.join()
    print([res.get() for res in res_l]) #拿到的是get_page的結果,其實完全沒必要拿該結果,該結果已經傳給回調函數處理了

'''
打印結果:
<進程3388> get https://www.baidu.com
<進程3389> get https://www.python.org
<進程3390> get https://www.openstack.org
<進程3388> get https://help.github.com/
<進程3387> parse https://www.baidu.com
<進程3389> get http://www.sina.com.cn/
<進程3387> parse https://www.python.org
<進程3387> parse https://help.github.com/
<進程3387> parse http://www.sina.com.cn/
<進程3387> parse https://www.openstack.org
[{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]
'''

示例

from multiprocessing import Pool
import time,random
import requests
import re

def get_page(url,pattern):
    response=requests.get(url)
    if response.status_code == 200:
        return (response.text,pattern)

def parse_page(info):
    page_content,pattern=info
    res=re.findall(pattern,page_content)
    for item in res:
        dic={
            'index':item[0],
            'title':item[1],
            'actor':item[2].strip()[3:],
            'time':item[3][5:],
            'score':item[4]+item[5]

        }
        print(dic)
if __name__ == '__main__':
    pattern1=re.compile(r'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S)

    url_dic={
        'http://maoyan.com/board/7':pattern1,
    }

    p=Pool()
    res_l=[]
    for url,pattern in url_dic.items():
        res=p.apply_async(get_page,args=(url,pattern),callback=parse_page)
        res_l.append(res)

    for i in res_l:
        i.get()

    # res=requests.get('http://maoyan.com/board/7')
    # print(re.findall(pattern,res.text))

爬蟲案例

如果在主進程中等待進程池中所有任務都執行完畢后，再統一處理結果，則無需回調函數

from multiprocessing import Pool
import time,random,os

def work(n):
    time.sleep(1)
    return n**2
if __name__ == '__main__':
    p=Pool()

    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,))
        res_l.append(res)

    p.close()
    p.join() #等待進程池中所有進程執行完畢

    nums=[]
    for res in res_l:
        nums.append(res.get()) #拿到所有結果
    print(nums) #主進程拿到所有的處理結果,可以在主進程中進行統一進行處理

示例

Some Content From——http://www.cnblogs.com/linhaifeng/articles/7430066.html

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 多進程並發編程 Python並發編程-多進程進程鎖 python多進程並發 Python並發編程系列之多進程（multiprocessing） python 3 並發編程之多進程 multiprocessing模塊 python並發編程之多進程(實踐篇) Python3 並發編程之多進程(二) python並發編程之多進程 python並發編程之多進程 python並發編程之多進程理論部分