一、數據共享
1.進程間的通信應該盡量避免共享數據的方式
2.進程間的數據是獨立的，可以借助隊列或管道實現通信，二者都是基於消息傳遞的。
雖然進程間數據獨立，但可以用過Manager實現數據共享，事實上Manager的功能遠不止於此。
|
1
2
3
4
|
命令就是一個程序，按回車就會執行（這個只是在windows情況下）
tasklist 查看進程
tasklist | findstr pycharm
#（findstr是進行過濾的），|就是管道(tasklist執行的內容就放到管道里面了，
管道后面的findstr pycharm就接收了)
|
3.（IPC）進程之間的通信有兩種實現方式：管道和隊列
from multiprocessing import Manager,Process,Lock
2 def work(dic,mutex):
3 # mutex.acquire()
4 # dic['count']-=1
5 # mutex.release()
6 # 也可以這樣加鎖
7 with mutex:
8 dic['count'] -= 1
9 if __name__ == '__main__':
10 mutex = Lock()
11 m = Manager() #實現共享，由於字典是共享的字典，所以得加個鎖
12 share_dic = m.dict({'count':100})
13 p_l = []
14 for i in range(100):
15 p = Process(target=work,args=(share_dic,mutex))
16 p_l.append(p) #先添加進去
17 p.start()
18 for i in p_l:
19 i.join()
20 print(share_dic)
21 # 共享就意味着會有競爭，
二、進程池
在利用Python進行系統管理的時候，特別是同時操作多個文件目錄，或者遠程控制多台主機，並行操作可以節約大量的時間。多進程是實現並發的手段之一，需要注意的問題是：
- 很明顯需要並發執行的任務通常要遠大於核數
- 一個操作系統不可能無限開啟進程，通常有幾個核就開幾個進程
- 進程開啟過多，效率反而會下降（開啟進程是需要占用系統資源的，而且開啟多余核數目的進程也無法做到並行）
例如當被操作對象數目不大時，可以直接利用multiprocessing中的Process動態成生多個進程，十幾個還好，但如果是上百個，上千個。。。手動的去限制進程數量卻又太過繁瑣，此時可以發揮進程池的功效。
那么什么是進程池呢？進程池就是控制進程數目
ps：對於遠程過程調用的高級應用程序而言，應該使用進程池，Pool可以提供指定數量的進程，供用戶調用，當有新的請求提交到pool中時，如果池還沒有滿，那么就會創建一個新的進程用來執行該請求；但如果池中的進程數已經達到規定最大值，那么該請求就會等待，直到池中有進程結束，就重用進程池中的進程。
進程池的結構：
創建進程池的類：如果指定numprocess為3，則進程池會從無到有創建三個進程，然后自始至終使用這三個進程去執行所有任務，不會開啟其他進程
1.創建進程池
|
1
|
Pool([numprocess [,initializer [, initargs]]]):創建進程池
|
2.參數介紹
|
1
2
3
|
numprocess:要創建的進程數，如果省略，將默認為cpu_count()的值,可os.cpu_count()查看
initializer：是每個工作進程啟動時要執行的可調用對象，默認為
None
initargs：是要傳給initializer的參數組
|
3.方法介紹
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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():關閉進程池，防止進一步操作。禁止往進程池內在添加任務（需要注意的是一定要寫在close()的上方）
|
|
1
|
P.jion():等待所有工作進程退出。此方法只能在close（）或teminate()之后調用
|
應用1：
from multiprocessing import Pool
2 import os,time
3 def task(n):
4 print('[%s] is running'%os.getpid())
5 time.sleep(2)
6 print('[%s] is done'%os.getpid())
7 return n**2
8 if __name__ == '__main__':
9 # print(os.cpu_count()) #查看cpu個數
10 p = Pool(4) #最大四個進程
11 for i in range(1,7):#開7個任務
12 res = p.apply(task,args=(i,)) #同步的，等着一個運行完才執行另一個
13 print('本次任務的結束：%s'%res)
14 p.close()#禁止往進程池內在添加任務
15 p.join() #在等進程池
16 print('主')
# ----------------
2 # 那么我們為什么要用進程池呢？這是因為進程池使用來控制進程數目的，
3 # 我們需要幾個就開幾個進程。如果不用進程池實現並發的話，會開很多的進程
4 # 如果你開的進程特別多，那么你的機器就會很卡，所以我們把進程控制好，用幾個就
5 # 開幾個，也不會太占用內存
6 from multiprocessing import Pool
7 import os,time
8 def walk(n):
9 print('task[%s] running...'%os.getpid())
10 time.sleep(3)
11 return n**2
12 if __name__ == '__main__':
13 p = Pool(4)
14 res_obj_l = []
15 for i in range(10):
16 res = p.apply_async(walk,args=(i,))
17 # print(res) #打印出來的是對象
18 res_obj_l.append(res) #那么現在拿到的是一個列表，怎么得到值呢？我們用個.get方法
19 p.close() #禁止往進程池里添加任務
20 p.join()
21 # print(res_obj_l)
22 print([obj.get() for obj in res_obj_l]) #這樣就得到了
那么什么是同步，什么是異步呢？
同步就是指一個進程在執行某個請求的時候，若該請求需要一段時間才能返回信息，那么這個進程將會一直等待下去，直到收到返回信息才繼續執行下去
異步是指進程不需要一直等下去，而是繼續執行下面的操作，不管其他進程的狀態。當有消息返回時系統會通知進程進行處理，這樣可以提高執行的效率。
什么是串行，什么是並行呢？
舉例：能並排開幾輛車的就可以說是“並行”，只能一輛一輛開的就屬於“串行”了。很明顯，並行的速度要比串行的快得多。（並行互不影響，串行的等着一個完了才能接着另一個）
應用2：
使用進程池維護固定數目的進程（以前客戶端和服務端的改進）
from socket import *
2 from multiprocessing import Pool
3 s = socket(AF_INET,SOCK_STREAM)
4 s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) #端口重用
5 s.bind(('127.0.0.1',8081))
6 s.listen(5)
7 print('start running...')
8 def talk(coon,addr):
9 while True: # 通信循環
10 try:
11 cmd = coon.recv(1024)
12 print(cmd.decode('utf-8'))
13 if not cmd: break
14 coon.send(cmd.upper())
15 print('發送的是%s'%cmd.upper().decode('utf-8'))
16 except Exception:
17 break
18 coon.close()
19 if __name__ == '__main__':
20 p = Pool(4)
21 while True:#鏈接循環
22 coon,addr = s.accept()
23 print(coon,addr)
24 p.apply_async(talk,args=(coon,addr))
25 s.close()
26 #因為是循環，所以就不用p.join了
from socket import *
2 c = socket(AF_INET,SOCK_STREAM)
3 c.connect(('127.0.0.1',8081))
4 while True:
5 cmd = input('>>:').strip()
6 if not cmd:continue
7 c.send(cmd.encode('utf-8'))
8 data = c.recv(1024)
9 print('接受的是%s'%data.decode('utf-8'))
10 c.close()
三、回調函數
|
1
2
3
4
5
6
7
|
回調函數什么時候用？（回調函數在爬蟲中最常用）
造數據的非常耗時
處理數據的時候不耗時
你下載的地址如果完成了，就自動提醒讓主進程解析
誰要是好了就通知解析函數去解析（回調函數的強大之處）
|
需要回調函數的場景：進程池中任何一個任務一旦處理完了，就立即告知主進程：我好了額，你可以處理我的結果了。主進程則調用一個函數去處理該結果，該函數即回調函數
我們可以把耗時間（阻塞）的任務放到進程池中，然后指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，直接拿到的是任務的結果。
from multiprocessing import Pool
2 import requests
3 import os
4 import time
5 def get_page(url):
6 print('<%s> is getting [%s]' %(os.getpid(),url))
7 response = requests.get(url) #得到地址
8 time.sleep(2)
9 print('<%s> is done [%s]'%(os.getpid(),url))
10 return {'url':url,'text':response.text}
11 def parse_page(res):
12 '''解析函數'''
13 print('<%s> parse [%s]'%(os.getpid(),res['url']))
14 with open('db.txt','a') as f:
15 parse_res = 'url:%s size:%s\n' %(res['url'],len(res['text']))
16 f.write(parse_res)
17 if __name__ == '__main__':
18 p = Pool(4)
19 urls = [
20 'https://www.baidu.com',
21 'http://www.openstack.org',
22 'https://www.python.org',
23 'https://help.github.com/',
24 'http://www.sina.com.cn/'
25 ]
26 for url in urls:
27 obj = p.apply_async(get_page,args=(url,),callback=parse_page)
28 p.close()
29 p.join()
30 print('主',os.getpid()) #都不用.get()方法了
如果在主進程中等待進程池中所有任務都執行完畢后，再統一處理結果，則無需回調函數
from multiprocessing import Pool
2 import requests
3 import os
4 def get_page(url):
5 print('<%os> get [%s]' %(os.getpid(),url))
6 response = requests.get(url) #得到地址 response響應
7 return {'url':url,'text':response.text}
8 if __name__ == '__main__':
9 p = Pool(4)
10 urls = [
11 'https://www.baidu.com',
12 'http://www.openstack.org',
13 'https://www.python.org',
14 'https://help.github.com/',
15 'http://www.sina.com.cn/'
16 ]
17 obj_l= []
18 for url in urls:
19 obj = p.apply_async(get_page,args=(url,))
20 obj_l.append(obj)
21 p.close()
22 p.join()
23 print([obj.get() for obj in obj_l])