這里所謂的低級，高級是指封裝抽象的程度。
低級指os.fork()
高級是指 multiprocessing包
一般根據業務需求，一個主進程負責維護接收， 不同的子進程處理不同的需求。根據各同需求組合
多進程
多線程
多進程＋多線程
協程
也可基于uvloop事件啟動方式
低級版
def main_process():
r = os.fork()
if r == 0 :
print("sub_process_buiness")
else:
print("main process buiness")
高級版
def sub_process_write_data():
pass
def sub_process_read_date():
pass
def main_handler():
from multiprocessing import Process
w1=Process(target=sub_process_write_data)
w1.start()
w1.join()
r1=Process(target=sub_process_read_data)
r1.start()
r1.join()
if __name__ == "__main__":
main_handler()
以下部分轉自
作者：曉可加油
鏈接：https://www.jianshu.com/p/414e89248285
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。
Num01-->進程的創建-fork
Python的os模塊封裝了常見的系統調用，其中就包括fork，可以在Python程序中輕松創建子進程。
import os
# 注意，fork函數，只在Unix/Linux/Mac上運行，windows不可以
pid = os.fork()
if pid == 0:
print('我是子進程')
else:
print('我是父進程')
以上代碼加以說明如下：
程序執行到os.fork()時，操作系統會創建一個新的進程(子進程)，然后復制父進程的所有信息到子進程中。
然后父進程和子進程都會從fork()函數中得到一個返回值，在子進程中這個值一定是0，而父進程中是子進程的 id號。
在Unix/Linux操作系統中，提供了一個fork()系統函數，它非常特殊。
普通的函數調用，調用一次，返回一次，但是fork()調用一次，返回兩次，因為操作系統自動把當前進程(稱為父進程)復制了一份(稱為子進程)，然后，分別在父進程和子進程內返回。
子進程永遠返回0，而父進程返回子進程的ID。
這樣做的理由是，一個父進程可以fork出很多子進程，所以，父進程要記下每個子進程的ID，而子進程只需要調用getppid()就可以拿到父進程的ID。
Num02-->多進程修改全局變量
``
coding=utf-8
import os
import time
num = 0
注意，fork函數，只在Unix/Linux/Mac上運行，windows不可以
pid = os.fork()
if pid == 0:
num+=1
print('我是子進程---num=%d'%num)
else:
time.sleep(1)
num+=1
print('我是父進程---num=%d'%num)
在多進程中個，每個進程中所有數據(包括全局變量)都各擁有一份，互不影響。
Num03-->多次fork問題
Test01-->fork兩次產生四個進程
coding=utf-8
import os
import time
注意，fork函數，只在Unix/Linux/Mac上運行，windows不可以
pid = os.fork()
if pid == 0:
print('我是第一次fork中的子進程')
else:
print('我是第一次fork中的父進程')
pid = os.fork()
if pid == 0:
print('我是第二次fork中的子進程')
else:
print('我是第二次fork中的父進程')
time.sleep(1)
Test02-->fork兩次產生三個進程
! /usr/bin/env python3
-- coding:utf-8 --
import os
import time
def sing():
print('--我是第一次fork的子進程--')
time.sleep(1)
def dance():
ppid = os.fork()
if ppid > 0:
print('--我是第二次fork的父進程--')
time.sleep(1)
elif ppid == 0:
print('--我是第二次fork的子進程--')
time.sleep(1)
def main():
pid = os.fork()
if pid > 0:
dance()
elif pid == 0:
sing()
if name == "main":
main()
Num04-->進程的第一種創建方式-multiprocessing
multiprocessing模塊提供了一個Process類來代表一個進程對象。
coding=utf-8
from multiprocessing import Process
import os
子進程要執行的代碼
def fun_proc(name):
print('子進程運行中，name= %s ,pid=%d' % (name, os.getpid()))
if name=='main':
print('父進程 %d' % os.getpid())
p = Process(target=fun_proc, args=('我是子進程',))
print('子進程將要執行')
p.start()
p.join()
print('子進程已結束')
對以上代碼加以說明：
1，用Process類創建子進程時，只需要傳入一個執行函數和函數的參數(是一個元組)。
2，調用start()方式啟動子進程。
3，join()方法可以等待子進程結束后再繼續往下運行，通常用于進程間的同步。
Test01-->Process的語法如下：
Process([group [, target [, name [, args [, kwargs]]]]])
target：表示這個進程實例所調用對象；
args：表示調用對象的位置參數元組；
kwargs：表示調用對象的關鍵字參數字典；
name：為當前進程實例的別名；
group：大多數情況下用不到，表示在哪個組；
Process類常用方法：
is_alive()：判斷進程實例是否還在執行；
join([timeout])：是否等待進程實例執行結束，或等待多少秒；
start()：啟動進程實例(創建子進程)；
run()：如果沒有給定target參數，對這個對象調用start()方法時，就將執行對象中的run()方法；
terminate()：不管任務是否完成，立即終止；
Process類常用屬性：
name：當前進程實例別名，默認為Process-N，N為從1開始遞增的整數；
pid：當前進程實例的PID值；
Test02-->創建一個進程對象
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date : 2017-04-25 16:36:47
# @Author : xiaoke
from multiprocessing import Process
import os
from time import sleep
# 子進程要執行的代碼
def fun_proc(name, age, **kwargs):
for i in range(5):
print('子進程運行中，name= %s,age=%d ,pid=%d...' % (name, age,os.getpid()))
print(kwargs)
sleep(1)
if __name__=='__main__':
print('父進程 %d' % os.getpid())
p = Process(target=fun_proc, args=('我是子進程',66), kwargs={"得分":666})
print('子進程將要執行')
p.start()
sleep(1)
# p.terminate()# 提前結束子進程，不管子進程的任務是否完成
p.join()
print('子進程已結束')
# 結果如下：
# 父進程 7744
# 子進程將要執行
# 子進程運行中，name= 我是子進程,age=66 ,pid=8064...
# {'得分': 666}
# 子進程運行中，name= 我是子進程,age=66 ,pid=8064...
# {'得分': 666}
# 子進程運行中，name= 我是子進程,age=66 ,pid=8064...
# {'得分': 666}
# 子進程運行中，name= 我是子進程,age=66 ,pid=8064...
# {'得分': 666}
# 子進程運行中，name= 我是子進程,age=66 ,pid=8064...
# {'得分': 666}
# 子進程已結束
Test03-->創建兩個進程對象
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date : 2017-04-25 16:36:47
# @Author : xiaoke
#coding=utf-8
from multiprocessing import Process
import time
import os
def worker_1(interval):
print("worker_1,父進程(%s),當前進程(%s)"%(os.getppid(),os.getpid()))
t_start = time.time()
time.sleep(interval) #程序將會被掛起interval秒
t_end = time.time()
print("worker_1,執行時間為'%0.2f'秒"%(t_end - t_start))
def worker_2(interval):
print("worker_2,父進程(%s),當前進程(%s)"%(os.getppid(),os.getpid()))
t_start = time.time()
time.sleep(interval)
t_end = time.time()
print("worker_2,執行時間為'%0.2f'秒"%(t_end - t_start))
def main():
#輸出當前程序的ID
print("進程ID：%s"%os.getpid())
#創建兩個進程對象，target指向這個進程對象要執行的對象名稱，
#如果不指定name參數，默認的進程對象名稱為Process-N，N為一個遞增的整數
p1=Process(target=worker_1,args=(2,))
p2=Process(target=worker_2,name="xiaoke",args=(1,))
#使用"進程對象名稱.start()"來創建并執行一個子進程，
#這兩個進程對象在start后，就會分別去執行worker_1和worker_2方法中的內容
p1.start()
p2.start()
#同時父進程仍然往下執行，如果p2進程還在執行，將會返回True
print("p2.is_alive=%s"%p2.is_alive())
print("p1.is_alive=%s"%p1.is_alive())
#輸出p1和p2進程的別名和pid
print("--p1進程的別名和pid--")
print("p1.name=%s"%p1.name)
print("p1.pid=%s"%p1.pid)
print("--p2進程的別名和pid--")
print("p2.name=%s"%p2.name)
print("p2.pid=%s"%p2.pid)
#join括號中不攜帶參數，表示父進程在這個位置要等待p1進程執行完成后，再繼續執行下面的語句，一般用于進程間的數據同步，
# 如果不寫這一句，下面的is_alive判斷將會是True，
#可以嘗試著將下面的這條語句改成p1.join(1)，
#因為p2需要2秒以上才可能執行完成，父進程等待1秒很可能，不能讓p1完全執行完成，
#所以下面的print會輸出True，即p1仍然在執行
print("--p1進程是否執行完畢？？--")
p1.join()
print("p1.is_alive=%s"%p1.is_alive())
p2.join()
print("p2.is_alive=%s"%p2.is_alive())
if __name__ == '__main__':
main()
# 結果如下：
# 進程ID：4004
# p2.is_alive=True
# p1.is_alive=True
# --p1進程的別名和pid--
# p1.name=Process-1
# p1.pid=3352
# --p2進程的別名和pid--
# p2.name=xiaoke
# p2.pid=6092
# --p1進程是否執行完畢？？--
# worker_2,父進程(4004),當前進程(6092)
# worker_2,執行時間為'1.00'秒
# worker_1,父進程(4004),當前進程(3352)
# worker_1,執行時間為'2.00'秒
# p1.is_alive=False
# p2.is_alive=False
Num05-->進程的第二種創建方式--自己創建一個類，繼承Process類
定義：創建新的進程還可以使用類的方式。可以自定義一個類，繼承Process類。每次實例化這個類的時候，就等同于實例化這個進程對象。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date : 2017-04-25 16:36:47
# @Author : xiaoke
from multiprocessing import Process
import time
import os
#繼承Process類
class Process_Class(Process):
#因為Process類本身也有__init__方法，這個子類相當于重寫了這個方法，
#但這樣就會帶來一個問題，我們并沒有完全的初始化一個Process類，
# 所以就不能使用從這個類繼承的一些方法和屬性，
#最好的方法就是將繼承類本身傳遞給Process.__init__方法，完成這些初始化操作
def __init__(self,interval):
Process.__init__(self)
# 傳遞進來的屬性
self.interval = interval
#重寫了Process類的run()方法
def run(self):
print("子進程(%s) 開始執行，父進程為(%s)"%(os.getpid(),os.getppid()))
t_start = time.time()
time.sleep(self.interval)
t_stop = time.time()
print("子進程(%s)執行結束，耗時%0.2f秒"%(os.getpid(),t_stop-t_start))
if __name__=="__main__":
t_start = time.time()
print("當前程序進程(%s)"%os.getpid())
p1 = Process_Class(2)
#對一個不包含target屬性的Process類執行start()方法，就會運行這個類中的run()方法，所以這里會執行p1.run()
p1.start()
p1.join()
t_stop = time.time()
print("父進程(%s)執行結束，耗時%0.2f秒"%(os.getpid(),t_stop-t_start))
# 結果為：
# 當前程序進程(14736)
# 子進程(4292) 開始執行，父進程為(14736)
# 子進程(4292)執行結束，耗時2.00秒
# 父進程(14736)執行結束，耗時2.11秒
Num06-->進程池--Pool
當需要創建的子進程數量不多時，可以直接利用multiprocessing中的Process動態生成多個進程。但如果是上百甚至上千個目標，手動的去創建進程的工作量巨大，此時就可以用到multiprocessing模塊提供的Pool方法。
初始化Pool時，可以指定一個最大進程數，當有新的請求提交到Pool中時，如果池還沒有滿，那么就會創建一個新的進程用來執行該請求；但如果池中的進程數已經達到指定的最大值，那么該請求就會等待，直到池中有進程結束，才會創建新的進程來執行。
#采用非阻塞的方式
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/4/27 8:35
# @Author : xiaoke
# @Site :
# @File : Test33.py
# @Software: PyCharm Community Edition
from multiprocessing import Pool
import os, time, random
def worker(num):
t_start = time.time()
print("子進程-%s開始執行,進程號為%d" % (num, os.getpid()))
# random.random()隨機生成0~1之間的浮點數
time.sleep(random.random() * 2)
t_stop = time.time()
print("子進程-%s執行完畢，耗時%0.2f" % (num, t_stop - t_start))
def main():
# 定義一個進程池，最大進程數5
p = Pool(3)
for i in range(10):
# Pool.apply_async(要調用的目標,(傳遞給目標的參數元祖,))
# 異步的方式，每次循環將會用空閑出來的子進程去調用目標
p.apply_async(worker, (i,))
m_start = time.time()
print("----start----")
p.close() # 關閉進程池，關閉后p不再接收新的請求
p.join() # 等待p中所有子進程執行完成，必須放在close語句之后
print("-----end-----")
m_stop = time.time()
print("父進程執行完畢，耗時%0.2f" % (m_stop - m_start))
if __name__ == '__main__':
main()
# 結果如下：
# ----start----
# 子進程-0開始執行,進程號為3360
# 子進程-1開始執行,進程號為16084
# 子進程-2開始執行,進程號為12580
# 子進程-1執行完畢，耗時0.05
# 子進程-3開始執行,進程號為16084
# 子進程-3執行完畢，耗時0.80
# 子進程-4開始執行,進程號為16084
# 子進程-2執行完畢，耗時1.63
# 子進程-5開始執行,進程號為12580
# 子進程-0執行完畢，耗時1.80
# 子進程-6開始執行,進程號為3360
# 子進程-4執行完畢，耗時1.55
# 子進程-7開始執行,進程號為16084
# 子進程-7執行完畢，耗時0.07
# 子進程-8開始執行,進程號為16084
# 子進程-8執行完畢，耗時0.05
# 子進程-9開始執行,進程號為16084
# 子進程-5執行完畢，耗時1.01
# 子進程-6執行完畢，耗時1.10
# 子進程-9執行完畢，耗時1.57
# -----end-----
# 父進程執行完畢，耗時4.26
multiprocessing.Pool常用函數解析：
apply_async(func[, args[, kwds]]) ：使用非阻塞方式調用func(并行執行，堵塞方式必須等待上一個進程退出才能執行下一個進程)，args為傳遞給func的參數列表，kwds為傳遞給func的關鍵字參數列表；
apply(func[, args[, kwds]])：使用阻塞方式調用func
close()：關閉Pool，使其不再接受新的任務；
terminate()：不管任務是否完成，立即終止；
join()：主進程阻塞，等待子進程的退出， 必須在close或terminate之后使用；
采用apply阻塞的方式
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/4/27 8:35
# @Author : xiaoke
# @Site :
# @File : Test33.py
# @Software: PyCharm Community Edition
from multiprocessing import Pool
import os, time, random
def worker(num):
t_start = time.time()
print("子進程-%s開始執行,進程號為%d" % (num, os.getpid()))
# random.random()隨機生成0~1之間的浮點數
time.sleep(random.random() * 2)
t_stop = time.time()
print("子進程-%s執行完畢，耗時%0.2f" % (num, t_stop - t_start))
def main():
# 定義一個進程池，最大進程數5
p = Pool(3)
for i in range(10):
# Pool.apply_async(要調用的目標,(傳遞給目標的參數元祖,))
# 異步的方式，每次循環將會用空閑出來的子進程去調用目標
p.apply(worker, (i,))
m_start = time.time()
print("----start----")
p.close() # 關閉進程池，關閉后p不再接收新的請求
p.join() # 等待p中所有子進程執行完成，必須放在close語句之后
print("-----end-----")
m_stop = time.time()
print("父進程執行完畢，耗時%0.2f" % (m_stop - m_start))
if __name__ == '__main__':
main()
# 結果如下：
# 子進程-0開始執行,進程號為4464
# 子進程-0執行完畢，耗時1.75
# 子進程-1開始執行,進程號為11640
# 子進程-1執行完畢，耗時1.33
# 子進程-2開始執行,進程號為8756
# 子進程-2執行完畢，耗時1.86
# 子進程-3開始執行,進程號為4464
# 子進程-3執行完畢，耗時0.70
# 子進程-4開始執行,進程號為11640
# 子進程-4執行完畢，耗時1.29
# 子進程-5開始執行,進程號為8756
# 子進程-5執行完畢，耗時0.69
# 子進程-6開始執行,進程號為4464
# 子進程-6執行完畢，耗時0.33
# 子進程-7開始執行,進程號為11640
# 子進程-7執行完畢，耗時1.83
# 子進程-8開始執行,進程號為8756
# 子進程-8執行完畢，耗時1.58
# 子進程-9開始執行,進程號為4464
# 子進程-9執行完畢，耗時1.37
# ----start----
# -----end-----
# 父進程執行完畢，耗時0.08
Num07-->進程間的通信--Queue
進程(Process)之間有時間需要通信，操作系統提供了很多機制來實現進程間的通信。如Queue、Pipes等。Queue本身是一個消息隊列。
Test01--> 先看一個簡單的案例：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/4/27 8:35
# @Author : xiaoke
# @Site :
# @File : Test33.py
# @Software: PyCharm Community Edition
from multiprocessing import Queue
q = Queue(3) # 初始化一個Queue對象，最多可接收三條put消息
q.put("消息1")
q.put("消息2")
print(q.full()) # False
q.put("消息3")
print(q.full()) # True
# 因為消息列隊已滿下面的try都會拋出異常，第一個try會等待3秒后再拋出異常，第二個Try會立刻拋出異常
try:
q.put("消息4", True, 3)
except:
print("消息列隊已滿，現有消息數量:%s" % q.qsize())
try:
q.put_nowait("消息4")
except:
print("消息列隊已滿，現有消息數量:%s" % q.qsize())
# 推薦的方式，先判斷消息列隊是否已滿，再寫入
if not q.full():
q.put_nowait("消息4")
# 讀取消息時，先判斷消息列隊是否為空，再讀取
if not q.empty():
for i in range(q.qsize()):
# print("取出消息：%s" % q.get())
print("取出消息：%s" % q.get_nowait())
# 結果是：
# False
# True
# 消息列隊已滿，現有消息數量:3
# 消息列隊已滿，現有消息數量:3
# 取出消息：消息1
# 取出消息：消息2
# 取出消息：消息3
以上代碼加以說明：
初始化Queue()對象時(例如：q=Queue())，若括號中沒有指定最大可接收的消息數量，或數量為負值，那么就代表可接受的消息數量沒有上限(直到內存的盡頭)；
Queue.qsize()：返回當前隊列包含的消息數量；
Queue.empty()：如果隊列為空，返回True，反之False ；
Queue.full()：如果隊列滿了，返回True,反之False；
Queue.get([block[, timeout]])：獲取隊列中的一條消息，然后將其從列隊中移除，block默認值為True；
1)如果block使用默認值，且沒有設置timeout(單位秒)，消息列隊如果為空，此時程序將被阻塞(停在讀取狀態)，直到從消息列隊讀到消息為止，如果設置了timeout，則會等待timeout秒，若還沒讀取到任何消息，則拋出"Queue.Empty"異常；
2)如果block值為False，消息列隊如果為空，則會立刻拋出"Queue.Empty"異常；
Queue.get_nowait()：相當Queue.get(False)；
Queue.put(item,[block[, timeout]])：將item消息寫入隊列，block默認值為True；
1)如果block使用默認值，且沒有設置timeout(單位秒)，消息列隊如果已經沒有空間可寫入，此時程序將被阻塞(停在寫入狀態)，直到從消息列隊騰出空間為止，如果設置了timeout，則會等待timeout秒，若還沒空間，則拋出"Queue.Full"異常；
2)如果block值為False，消息列隊如果沒有空間可寫入，則會立刻拋出"Queue.Full"異常；
Queue.put_nowait(item)：相當Queue.put(item, False)；
Test02-->在父進程創建兩個子進程，一個往Queue里面寫數據，一個從Queue里面讀數據。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/4/27 8:35
# @Author : xiaoke
# @Site :
# @File : Test33.py
# @Software: PyCharm Community Edition
from multiprocessing import Process, Queue
import os, time, random
# 寫數據進程
def write(q):
for value in ['A', 'B', 'C', 'quit']:
print('Put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
# 讀數據進程
def read(q):
while True:
if not q.empty():
value = q.get(True)
print('Get %s from queue.' % value)
# 因為讀進程是一個死循環，所以要設置一個標記，用于退出
if value == "quit":
break
time.sleep(random.random())
if __name__ == '__main__':
# 父進程創建Queue，并傳給各個子進程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 啟動子進程pw，寫入:
pw.start()
# 等待pw結束:
pw.join()
# 啟動子進程pr，讀取:
pr.start()
pr.join()
print('所有數據都寫入并且讀完')
print("pw is alive:%s" % pw.is_alive())
print("pr is alive:%s" % pr.is_alive())
# 結果如下：
# Put A to queue...
# Put B to queue...
# Put C to queue...
# Put quit to queue...
# Get A from queue.
# Get B from queue.
# Get C from queue.
# Get quit from queue.
# 所有數據都寫入并且讀完
# pw is alive:False
# pr is alive:False
Test03-->進程池Pool中的Queue來進行進程間的通信
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author : xiaoke
# @Site :
# @File : Test33.py
# @Software: PyCharm Community Edition
from multiprocessing import Pool, Queue, Manager
import os
import time
import random
def task_write(q):
for s in ('hello', 'python', 'world', 'quit'):
q.put(s) # 向隊列中添加消息
print('%s 進程向隊列中添加消息:%s' % (os.getpid(), s))
time.sleep(random.random() * 2)
print('%s 進程要結束了' % os.getpid())
def task_read(q):
while True:
msg = q.get() # 阻塞式從隊列中收消息
print('%s 進程從隊列中取出消息:%s' % (os.getpid(), msg))
if msg == "quit":
break
time.sleep(random.random() * 2)
print('%s 進程要結束了' % os.getpid())
def main():
# 1.創建消息隊列對象
# q = Queue() #只能用于父子進程
# Manger().Queue() 消息隊列可用于進程池
q = Manager().Queue()
# 2.創建進程池，里面放兩個進程
my_pool = Pool(2)
# 3.添加任務
# 采用阻塞的方式
my_pool.apply(task_write, args=(q,))
my_pool.apply(task_read, args=(q,))
# 采用非阻塞的方式
# my_pool.apply_async(task_write, args=(q,))
# my_pool.apply_async(task_read, args=(q,))
# 4.關閉進程池
my_pool.close()
# 5.等待所有進程結束
my_pool.join()
if __name__ == "__main__":
main()
# 采用非阻塞的方式結果：
# 7380 進程向隊列中添加消息:hello
# 11256 進程從隊列中取出消息:hello
# 7380 進程向隊列中添加消息:python
# 11256 進程從隊列中取出消息:python
# 7380 進程向隊列中添加消息:world
# 11256 進程從隊列中取出消息:world
# 7380 進程向隊列中添加消息:quit
# 11256 進程從隊列中取出消息:quit
# 11256 進程要結束了
# 7380 進程要結束了
# 采用阻塞的方式結果：
# 96 進程向隊列中添加消息:hello
# 96 進程向隊列中添加消息:python
# 96 進程向隊列中添加消息:world
# 96 進程向隊列中添加消息:quit
# 96 進程要結束了
# 12412 進程從隊列中取出消息:hello
# 12412 進程從隊列中取出消息:python
# 12412 進程從隊列中取出消息:world
# 12412 進程從隊列中取出消息:quit
# 12412 進程要結束了
)
