python 程序實現多工
1.匯入程序模組
import multiprocessing
1.2 Process程序類的語法結構如下:
Process([group [, target [, name [, args [, kwargs]]]]])
- group:指定程序組,目前只能使用None
- target:執行的目標任務名
- name:程序名字
- args:以元組方式給執行任務傳參
- kwargs:以字典方式給執行任務傳參
Process建立的例項物件的常用方法:
- start():啟動子程序例項(建立子程序)
- join([timeout]):是否等待子程序執行結束,或等待多少秒
- terminate():不管任務是否完成,立即終止子程序
Process建立的例項物件的常用屬性:
- name:當前程序的別名,預設為Process-N,N為從1開始遞增的整數
- pid:當前程序的pid(程序號)multiprocessing.current_process().pid 或os.getpid()
- 獲取父程序編號:os.getppid()
2.程序注意點
- 程序之間不共享全域性變數(建立子程序其實是對主程序進行拷貝,程序程序之間相互獨立,訪問的全域性變數不是同一個)需要使用Queue
- 主程序預設會等待所有的子程序執行完成後程式才會退出
- 可以為設定守護主程序 ,但是隻能通過sub_process.daemon = True
- 可以直接銷燬子程序,主程序退出之前把子程序銷燬,sub_process.terminate()
3.程序間通訊
queue新增資料:
queue = multiprocessing.Queue(maxsize=-1) # -1表示新增任意多個數據
queue.put(資料) # queue可以放入任何型別的資料 ,如果佇列滿了,資料會進行等待,直到訊息佇列有空閒位置才放入資料
queue.put_nowait(資料)# 佇列如果滿了,此方法會報錯
queue判斷佇列是否滿了
result = queue.full() # 若滿了返回True
result = queue.empty() # 坑點,此方法判斷訊息佇列不靠譜 ,不推薦使用
queue.qusize() # 檢視當前佇列裡訊息的個數
queue獲取資料
value = queue.get() # 佇列中的元素是FIFO(先進先出)
value = queue.get_nowait() # 若佇列為空立即報錯,不推薦使用
利用訊息佇列完成程序通訊
import multiprocessing
import time
# 新增資料到指定佇列
def add_data(current_queue):
# 迴圈往指定佇列裡面新增資料
for value in ["蘋果", "香蕉", "西瓜", "榴蓮", "山竹"]:
# 判斷佇列是否滿了
if current_queue.full():
print("佇列滿了")
break
current_queue.put(value)
print("add_data:", value)
time.sleep(0.2)
# 讀取資料從指定佇列
def read_data(current_queue):
while True:
#if current_queue.empty():
if current_queue.qsize() == 0:
print("佇列空了")
break
value = current_queue.get()
print("read_data:", value)
if __name__ == '__main__':
# 建立訊息佇列
queue = multiprocessing.Queue(3)
# 新增資料的程序
add_data_process = multiprocessing.Process(target=add_data, args=(queue,))
read_data_process = multiprocessing.Process(target=read_data, args=(queue,))
# 啟動程序執行對應的程式
add_data_process.start()
# 主程序等待新增資料的子程序執行完成以後再啟動讀取資料的程序
add_data_process.join()
read_data_process.start()
程序池
用來自動建立大量執行緒,process_pool = multiprocessing.Pool(max_size)
初始化程序池時,可以指定一個最大程序數,當有新的請求提交到Pool時,如果池內有空閒程序則使用該程序,若無空閒程序,且程序池還未達到最大值,那麼久會建立一個新的程序來執行該請求,若達到最大值,那麼該程序就會等待,直到池中有程序結束,才會使用之前的程序來執行新的任務。
程序池分為同步執行任務和非同步執行任務。
同步執行任務:process_pool.apply(work),預設建立的子程序為守護主程序的,但主程序會被阻塞到子程序執行結束
非同步執行任務:process_pool.apply_async(work),建立的子程序是守護主程序的,但它是非阻塞的。
主程序需要等待程序池把所有任務完成才能退出
pool.close() # 關閉程序池,明確告訴主程序不再接受其它任務
pool.join() # 等待程序池完成所有任務
demo 程序池完成多工版資料夾拷貝
import os
import shutil # 檔案操作的高階模組
import multiprocessing
import time
# 拷貝檔案的操作
def copy_work(src_dir_path, dst_dir_path, file_name):
src_file_path = src_dir_path + "/" + file_name # test/1.txt
dst_file_path = dst_dir_path + "/" + file_name
print(multiprocessing.current_process().pid)
# 開啟目標檔案
with open(dst_file_path, "wb") as dst_file:
# 開啟原始檔,把原始檔的資料寫入到目標檔案
with open(src_file_path, "rb") as src_file:
# 迴圈讀取檔案中的資料
while True:
# 讀取一次原始檔中的資料
src_file_data = src_file.read(1024)
if src_file_data:
# 把原始檔的資料寫入到目標檔案裡面
dst_file.write(src_file_data)
else:
break
time.sleep(0.5)
if __name__ == '__main__':
# 源目錄的路徑
src_dir_path = "test"
# 目標目錄的路徑
dst_dir_path = "/home/python/Desktop/test"
# 判斷目標目錄是否在桌面存在
if os.path.exists(dst_dir_path):
# 程式碼執行到此,說明桌面上有指定的目標目錄,遞迴刪除非空目錄
shutil.rmtree(dst_dir_path)
os.mkdir(dst_dir_path)
file_name_list = os.listdir(src_dir_path)
# 建立程序池
pool = multiprocessing.Pool(3)
for file_name in file_name_list:
# print(file_name)
# 使用程序池以非同步的方式完成多工檔案的拷貝
pool.apply_async(copy_work, args=(src_dir_path, dst_dir_path, file_name))
# 關閉程序池
pool.close()
pool.join()
程序池在執行任務的時候會盡量少建立程序,合理利用現有程序完成多工,這樣可以減少資源開銷。