多進程（multiprocessing module）

1、多進程

1.1 多進程的概念

　　因爲GIL的存在，python中的多線程其實並非真正的多線程，若是想要充分地使用多核CPU的資源，在python中大部分狀況須要使用多進程。Python提供了很是好用的多進程包multiprocessing，只須要定義一個函數，Python會完成其餘全部事情。藉助這個包，能夠輕鬆完成從單進程到併發執行的轉換。multiprocessing支持子進程、通訊和共享數據、執行不一樣形式的同步，提供了Process、Queue、Pipe、Lock等組件。

　　multiprocessing包是Python中的多進程管理包。與threading.Thread相似，它能夠利用multiprocessing.Process對象來建立一個進程。該進程能夠運行在Python程序內部編寫的函數。該Process對象與Thread對象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition類 (這些對象能夠像多線程那樣，經過參數傳遞給各個進程)，用以同步進程，其用法與threading包中的同名類一致。因此，multiprocessing的很大一部份與threading使用同一套API，只不過換到了多進程的情境。

但在使用這些共享API的時候，咱們要注意如下幾點:

　　a、在UNIX平臺上，當某個進程終結以後，該進程須要被其父進程調用wait，不然進程成爲殭屍進程(Zombie)。因此，有必要對每一個Process對象調用join()方法 (實際上等同於wait)。對於多線程來講，因爲只有一個進程，因此不存在此必要性。

　　b、Windows系統下，須要注意的是要想啓動一個子進程，必須加上 if __name__ == "__main__"，進程相關的要寫在這句下面。

1.2 建立進程的兩種方式

直接建立：

 1 from multiprocessing import Process
 2 import time
 3 def f(name):
 4     time.sleep(1)
 5     print('hello', name,time.ctime())
 6 
 7 if __name__ == '__main__':
 8     p_list=[]
 9     for i in range(3):
10         p = Process(target=f, args=('alvin',))
11         p_list.append(p)
12         p.start()
13     for i in p_list:
14         p.join()
15     print('end')

Demo1

類式調用：

 1 from multiprocessing import Process
 2 import time
 3 
 4 class MyProcess(Process):
 5     def __init__(self):
 6         super(MyProcess, self).__init__()
 7         #self.name = name
 8 
 9     def run(self):
10         time.sleep(1)
11         print ('hello', self.name,time.ctime())
12 
13 
14 if __name__ == '__main__':
15     p_list=[]
16     for i in range(3):
17         p = MyProcess()
18         p.start()
19         p_list.append(p)
20 
21     for p in p_list:
22         p.join()
23 
24     print('end')

Demo2

2、Process類

2.1 構造方法

　　 Process([group [, target [, name [, args [, kwargs]]]]])

　　group: 線程組，目前尚未實現，庫引用中提示必須是None； 
　　target: 要執行的方法； 
　　name: 進程名； 
　　args/kwargs: 要傳入方法的參數。

2.2 實例方法

　　 is_alive()：返回進程是否在運行。

　　join([timeout])：阻塞當前上下文環境的進程程，直到調用此方法的進程終止或到達指定的timeout（可選參數）。

　　start()：進程準備就緒，等待CPU調度

　　run()：strat()調用run方法，若是實例進程時未制定傳入target，這star執行t默認run()方法。

　　terminate()：無論任務是否完成，當即中止工做進程

2.3 屬性

　  　authkey

　　daemon：和線程的setDeamon功能同樣

　　exitcode(進程在運行時爲None、若是爲–N，表示被信號N結束）

　　name：進程名字。

　　pid：進程號。

 1 import time
 2 from  multiprocessing import Process
 3 
 4 def foo(i):
 5     time.sleep(1)
 6     print (p.is_alive(),i,p.pid)
 7     time.sleep(1)
 8 
 9 if __name__ == '__main__':
10     p_list=[]
11     for i in range(10):
12         p = Process(target=foo, args=(i,))
13         #p.daemon=True
14         p_list.append(p)
15 
16     for p in p_list:
17         p.start()
18     # for p in p_list:
19     #     p.join()
20 
21     print('main process end')

Demo

3、進程間通訊

不一樣進程間內存是不共享的，要想實現兩個進程間的數據交換，能夠用如下方法：

3.1 Queues 使用方法跟threading裏的queue相似 --> 將q做爲參數傳遞給子進程。

 1 from multiprocessing import Process, Queue
 2 
 3 def f(q,n):
 4     q.put([42, n, 'hello'])
 5 
 6 if __name__ == '__main__':
 7     q = Queue()
 8     p_list=[]
 9     for i in range(3):
10         p = Process(target=f, args=(q,i))
11         p_list.append(p)
12         p.start()
13     print(q.get())
14     print(q.get())
15     print(q.get())
16     for i in p_list:
17             i.join()

Demo1

3.2 Pipes --> 經過管道Pipe實現。

 1 from multiprocessing import Process, Pipe
 2  
 3 def f(conn):
 4     conn.send([42, None, 'hello'])
 5     conn.close()
 6  
 7 if __name__ == '__main__':
 8     parent_conn, child_conn = Pipe()
 9     p = Process(target=f, args=(child_conn,))
10     p.start()
11     print(parent_conn.recv())   # prints "[42, None, 'hello']"
12     p.join()

Demo2

3.3 數據共享（Manager）

Manager（）返回的管理器對象控制一個服務器進程，該進程保存Python對象並容許其餘進程使用代理操做它們。

 1 from multiprocessing import Process, Manager
 2 
 3 def f(d, l,n):
 4     d[n] = '1'
 5     d['2'] = 2
 6     d[0.25] = None
 7     l.append(n)
 8     print(l)
 9 
10 if __name__ == '__main__':
11     with Manager() as manager:
12         d = manager.dict()
13 
14         l = manager.list(range(5))
15         p_list = []
16         for i in range(10):
17             p = Process(target=f, args=(d, l,i))
18             p.start()
19             p_list.append(p)
20         for res in p_list:
21             res.join()
22 
23         print(d)
24         print(l)

Demo

3.4 進程同步

Without using the lock output from the different processes is liable to get all mixed up.　　若是不使用來自不一樣進程的鎖定輸出，則可能會混淆不清。

 1 from multiprocessing import Process, Lock
 2 
 3 def f(l, i):
 4     l.acquire()
 5     try:
 6         print('hello world', i)
 7     finally:
 8         l.release()
 9 
10 if __name__ == '__main__':
11     lock = Lock()
12 
13     for num in range(10):
14         Process(target=f, args=(lock, num)).start()

3.5 進程池

進程池內部維護一個進程序列，當使用時，則去進程池中獲取一個進程，若是進程池序列中沒有可供使用的進程，那麼程序就會等待，直到進程池中有可用進程爲止。

進程池中有兩個方法：

• apply
• apply_async

 1 from  multiprocessing import Process,Pool
 2 import time
 3  
 4 def Foo(i):
 5     time.sleep(2)
 6     return i+100
 7  
 8 def Bar(arg):
 9     print('-->exec done:',arg)
10  
11 pool = Pool(5)
12  
13 for i in range(10):
14     pool.apply_async(func=Foo, args=(i,),callback=Bar)
15     #pool.apply(func=Foo, args=(i,))
16  
17 print('end')
18 pool.close()
19 pool.join()

4、協程

協程，又稱微線程，纖程。英文名Coroutine。一句話說明什麼是線程：協程是一種用戶態的輕量級線程。

協程擁有本身的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其餘地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。所以：

協程能保留上一次調用時的狀態（即全部局部狀態的一個特定組合），每次過程重入時，就至關於進入上一次調用的狀態，換種說法：進入上一次離開時所處邏輯流的位置。

協程的好處：

• 無需線程上下文切換的開銷
• 無需原子操做鎖定及同步的開銷
• "原子操做(atomic operation)是不須要synchronized"，所謂原子操做是指不會被線程調度機制打斷的操做；這種操做一旦開始，就一直運行到結束，中間不會有任何 context switch （切換到另外一個線程）。原子操做能夠是一個步驟，也能夠是多個操做步驟，可是其順序是不能夠被打亂，或者切割掉只執行部分。視做總體是原子性的核心。
• 方便切換控制流，簡化編程模型
• 高併發+高擴展性+低成本：一個CPU支持上萬的協程都不是問題。因此很適合用於高併發處理。

 

缺點：

• 沒法利用多核資源：協程的本質是個單線程,它不能同時將 單個CPU 的多個核用上,協程須要和進程配合才能運行在多CPU上.固然咱們平常所編寫的絕大部分應用都沒有這個必要，除非是cpu密集型應用。
• 進行阻塞（Blocking）操做（如IO時）會阻塞掉整個程序

4.1 使用yield實現協程操做

 1 import time 
 2 import queue
 3 def consumer(name):  4 print("--->starting eating baozi...")  5 while True:  6 new_baozi = yield  7 print("[%s] is eating baozi %s" % (name,new_baozi))  8 #time.sleep(1)  9 10 def producer(): 11 12 r = con.__next__() 13 r = con2.__next__() 14 n = 0 15 while n < 5: 16 n +=1 17  con.send(n) 18  con2.send(n) 19 print("\033[32;1m[producer]\033[0m is making baozi %s" %n ) 20 21 22 if __name__ == '__main__': 23 con = consumer("c1")　　　　# 建立生成器對象 24 con2 = consumer("c2")　　　 # 建立生成器對象 25 p = producer()　　　　　　　　# 執行producer函數

協程標準定義，即符合什麼條件就能稱之爲協程：

1. 必須在只有一個單線程裏實現併發
2. 修改共享數據不需加鎖
3. 用戶程序裏本身保存多個控制流的上下文棧
4. 一個協程遇到IO操做自動切換到其它協程

基於上面這4點定義，咱們剛纔用yield實現的程並不能算是合格的線程，由於它有一點功能沒實現。

4.2 Greenlet

greenlet是一個用C實現的協程模塊，相比與python自帶的yield，它可使你在任意函數之間隨意切換，而不需把這個函數先聲明爲generator

 1 # -*- coding:utf-8 -*-
 2  
 3  
 4 from greenlet import greenlet
 5  
 6 def test1():
 7     print(12)
 8     gr2.switch()
 9     print(34)
10     gr2.switch()
11  
12 def test2():
13     print(56)
14     gr1.switch()
15     print(78)
16  
17 gr1 = greenlet(test1)
18 gr2 = greenlet(test2)
19 gr1.switch()

感受確實用着比generator還簡單了，但好像尚未解決一個問題，就是遇到IO操做，自動切換，並不對。

4.3 Gevent 

Gevent 是一個第三方庫，能夠輕鬆經過gevent實現併發同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程。 Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度。

 1 import gevent
 2  
 3 def func1():
 4     print('\033[31;1m李闖在跟海濤搞...\033[0m')
 5     gevent.sleep(2)
 6     print('\033[31;1m李闖又回去跟繼續跟海濤搞...\033[0m')
 7  
 8 def func2():
 9     print('\033[32;1m李闖切換到了跟海龍搞...\033[0m')
10     gevent.sleep(1)
11     print('\033[32;1m李闖搞完了海濤，回來繼續跟海龍搞...\033[0m')
12  
13  
14 gevent.joinall([
15     gevent.spawn(func1),
16     gevent.spawn(func2),
17     #gevent.spawn(func3),
18 ])

輸出：

李闖在跟海濤搞...
李闖切換到了跟海龍搞...
李闖搞完了海濤，回來繼續跟海龍搞...
李闖又回去跟繼續跟海濤搞...

5、補充

5.1 同步與異步的性能區別

 1 import gevent
 2  
 3 def task(pid):
 4     """
 5     Some non-deterministic task
 6     """
 7     gevent.sleep(0.5)
 8     print('Task %s done' % pid)
 9  
10 def synchronous():
11     for i in range(1,10):
12         task(i)
13  
14 def asynchronous():
15     threads = [gevent.spawn(task, i) for i in range(10)]
16     gevent.joinall(threads)
17  
18 print('Synchronous:')
19 synchronous()
20  
21 print('Asynchronous:')
22 asynchronous()

上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。 初始化的greenlet列表存放在數組threads中，此數組被傳給gevent.joinall 函數，後者阻塞當前流程，並執行全部給定的greenlet。執行流程只會在 全部greenlet執行完後纔會繼續向下走。

5.2 遇到IO阻塞時會自動切換任務（gevent 庫中的 monkey 方法）--> 補丁

　　------->可以最大程度監聽IO阻塞，提升效率。

 1 from gevent import monkey
 2 monkey.patch_all()
 3 
 4 import gevent
 5 from  urllib.request import urlopen
 6  
 7 def f(url):
 8     print('GET: %s' % url)
 9     resp = urlopen(url)
10     data = resp.read()
11     print('%d bytes received from %s.' % (len(data), url))
12  
13 gevent.joinall([
14         gevent.spawn(f, 'https://www.python.org/'),
15         gevent.spawn(f, 'https://www.yahoo.com/'),
16         gevent.spawn(f, 'https://github.com/'),
17 ])