python中的進程代碼
一 multiprocessing模塊介紹
python中的多線程沒法利用多核優點，若是想要充分地使用多核CPU的資源（os.cpu_count()查看），在python中大部分狀況須要使用多進程。Python提供了multiprocessing。
multiprocessing模塊用來開啓子進程，並在子進程中執行咱們定製的任務（好比函數），該模塊與多線程模塊threading的編程接口相似。python
multiprocessing模塊的功能衆多：支持子進程、通訊和共享數據、執行不一樣形式的同步，提供了Process、Queue、Pipe、Lock等組件。編程
須要再次強調的一點是：與線程不一樣，進程沒有任何共享狀態，進程修改的數據，改動僅限於該進程內。windows
二 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字符的字符串。這個鍵的用途是爲涉及網絡鏈接的底層進程間通訊提供安全性，這類鏈接只有在具備相同的身份驗證鍵時才能成功（瞭解便可）
注：p.pid和p.ppid分別查詢當前進程id和父進程id同：os.getpid（）和os.getppid()併發
建立進程：
注意：windows建立進程必定要寫在if __name__ == '__main__':的裏面，由於子進程會執行一遍整個文件，若是放在外面會無限建立子進程，爆棧app
第一種方式dom
def func():
print('child')
if __name__ == '__main__':
p = Process(target=func) # 建立一個進程對象
p.start() # 告訴操做系統我要建立一個進程，直到執行了start纔有進程
第二種方式：
import time
import os
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)
self.walk() # 只有這種纔是真正的進程本身調用walk
def walk(self):
print("子進程：", os.getpid())
if __name__ == '__main__':
p1 = Piao('egon')
p2 = Piao('alex')
p3 = Piao('wupeiqi')
p4 = Piao('yuanhao')
p1.start() # start會自動調用run
p2.start()
p3.start()
p4.start()
p1.walk() # 這樣寫至關於在主進程調用類中的方法，只有在類中run中寫纔是子進程本身調用
print('主線程')
join詳解：
正常狀況下父進程會等子進程結束後再結束進程（儘管父進程早已結束），可是子進程和父進程是異步的，若想父進程等子進程結束後再進行下一步則用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：主進程等，等待子進程結束
這裏就會有個優化問題，多個子進程能夠裝在列表裏，而後循環join
import time,os
from multiprocessing import Process
def desk(i):
time.sleep(1)
print('%d:子進程%d乾的事，父進程是：%d' %(i,os.getpid(),os.getppid()))
if __name__ == '__main__':
p_list = []
for i in range(10):
p = Process(target=desk, args=(i,))
p.start()
# p.join()
p_list.append(p)
for p in p_list:
p.join()
print('+++++++++++++++++++++++++++')
守護進程
在windows中，父進程會等子進程結束才結束，若讓父進程不等子進程而可自行結束（子進程也要跟着結束），則須要將子進程設置爲守護進程
主進程建立守護進程
其一：守護進程會在主進程代碼執行結束後就終止，注意不是主進程所有結束，而是主進程的代碼執行結束，守護進程不等子進程
其二：守護進程內沒法再開啓子進程,不然拋出異常：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('主')
進程鎖
進程之間數據不共享,可是共享同一套文件系統,因此訪問同一個文件,或同一個打印終端,是沒有問題的，而共享帶來的是競爭，競爭帶來的結果就是錯亂，如何控制，就是加鎖處理。
加鎖能夠保證多個進程修改同一塊數據時，同一時間只能有一個任務能夠進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。
acqurie（）和release（）中間的程序被鎖住，只有有鑰匙才能訪問
#由併發變成了串行,犧牲了運行效率,但避免了競爭
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()
加鎖：由併發變成了串行,犧牲了運行效率,但避免了競爭
信號量（至關於一把鎖上好幾把鑰匙，這幾把鑰匙的擁有者可同時訪問）
互斥鎖 同時只容許一個線程更改數據，而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
# e.is_set是判斷是否Flag是True
from multiprocessing import Process,Event
import time,random
def traffic_light(e):
while True:
if e.is_set():
time.sleep(3)
print('紅燈亮')
e.clear() # 綠變紅
else:
time.sleep(3)
print('綠燈亮')
e.set() # 紅變綠
def car(i,e):
e.wait() # 觸發事件
print("%s車經過"%i)
if __name__ == '__main__':
e = Event() # 生成事件對象
# 起一個進程，只執行紅綠燈轉換
p = Process(target=traffic_light, args=(e,))
p.start()
for i in range(100):
if i%6 == 0:
time.sleep(random.randint(1,3))
car_pro = Process(target=car, args=(i,e))
car_pro.start()
隊列
from multiprocessing import Process,Queue 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()同樣 #瞭解 1 q.cancel_join_thread():不會在進程退出時自動鏈接後臺線程。能夠防止join_thread()方法阻塞 2 q.close():關閉隊列，防止隊列中加入更多數據。調用此方法，後臺線程將繼續寫入那些已經入隊列但還沒有寫入的數據，但將在此方法完成時立刻關閉。若是q被垃圾收集，將調用此方法。關閉隊列不會在隊列使用者中產生任何類型的數據結束信號或異常。例如，若是某個使用者正在被阻塞在get()操做上，關閉生產者中的隊列不會致使get()方法返回錯誤。 3 q.join_thread()：鏈接隊列的後臺線程。此方法用於在調用q.close()方法以後，等待全部隊列項被消耗。默認狀況下，此方法由不是q的原始建立者的全部進程調用。調用q.cancel_join_thread方法能夠禁止這種行爲
實例
from multiprocessing import Process, Queue
def q_get(q):
print(q.get())
def q_put(q):
q.put('hello')
if __name__ == '__main__':
q = Queue()
p1 = Process(target=q_put, args=(q,))
p1.start()
p2 = Process(target=q_get, args=(q,))
p2.start()
生產者消費者模型：
生產者消費者模式是經過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通信，而經過阻塞隊列來進行通信，因此生產者生產完數據以後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列裏取，阻塞隊列就至關於一個緩衝區，平衡了生產者和消費者的處理能力。
基於隊列實現生產者消費者模型
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，不必定要由生產者發，主進程裏一樣能夠發，但主進程須要等生產者結束後才應該發送該信號
但上述解決方式，在有多個生產者和多個消費者時，咱們則須要用一個很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()
# 阻塞到q隊列中全部的項目均被q.task_done（）方法爲止
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方法會一直阻塞。若是同一Pipe對象生成的另一端已經關閉，那麼recv方法會拋出EOFError。
注意：
foo，son = Pipe（）
p = Process(target=func,args=((foo,son),))這種傳參進去的管道雙端不是同一Pipe對象，相似於形參，可能關一端，另外一端不受影響
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異常。所以在生產者中關閉管道不會有任何效果，付費消費者中也關閉了相同的管道端點。
共享數據
進程間通訊應該儘可能避免使用本節所講的共享數據的方式
進程間數據是獨立的，能夠藉助於隊列或管道實現通訊，兩者都是基於消息傳遞的
雖然進程間數據獨立，但能夠經過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,
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}
進程之間操做共享的數據
進程池
假如16核的cpu，那麼最多也就只能跑16個進程，若是跑100個進程，則按時間輪轉片來分，每一個進程分到的時間就不多，進程池簡單說就是開一個只有固定可執行進程數的池子，而後若是有100個任務，輪着用着這四個進程，這就避免了開過多進程形成的浪費
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.map():傳兩個參數，一個是須要執行的方法，一個是可迭代對象（range(100)）
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_async(work,args=(i,)) #同步運行,阻塞、直到本次任務執行完畢拿到res
res_l.append(res)
#異步apply_async用法：若是使用異步提交的任務，主進程須要使用jion，等待進程池內任務都處理完，而後能夠用get收集結果，不然，主進程結束，進程池可能還沒來得及執行，也就跟着一塊兒結束了
p.close() # 必定要close一下，防止再向進程池添加
p.join() # 異步進程池主進程不等子進程，因此要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
回調函數
須要回調函數的場景：進程池中任何一個任務一旦處理完了，就當即告知主進程：我好了額，你能夠處理個人結果了。主進程則調用一個函數去處理該結果，該函數即回調函數
咱們能夠把耗時間（阻塞）的任務放到進程池中，而後指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。
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))
- 1. 【python】python代碼的縮進
- 2. 使用pyrasite進行python進程調試，改變運行中進程的代碼
- 3. Python中的進程池與線程池(包含代碼) Python中的進程池與線程池
- 4. python編程代碼-python編程代碼
- 5. 進程代碼
- 6. 進程(代碼)
- 7. python工程中c++代碼的編譯
- 8. python--(十五步代碼學會進程)
- 9. python多進程代碼實現
- 10. python中的進程， 線程
- 更多相關文章...
- • Markdown 代碼 - Markdown 教程
- • Eclipse 代碼模板 - Eclipse 教程
- • Scala 中文亂碼解決
- • IntelliJ IDEA代碼格式化設置
-
每一个你不满意的现在，都有一个你没有努力的曾经。