快捷導(dǎo)航

python 進(jìn)程池pool使用詳解

更新時(shí)間：2020年10月15日 11:43:04 作者：小名叫小明

這篇文章主要介紹了python 進(jìn)程池pool使用的相關(guān)資料，幫助大家更好的理解和學(xué)習(xí)python進(jìn)程，感興趣的朋友可以了解下

和選用線程池來(lái)關(guān)系多線程類似，當(dāng)程序中設(shè)置到多進(jìn)程編程時(shí)，Python 提供了更好的管理多個(gè)進(jìn)程的方式，就是使用進(jìn)程池。

在利用 Python 進(jìn)行系統(tǒng)管理的時(shí)候，特別是同時(shí)操作多個(gè)文件目錄，或者遠(yuǎn)程控制多臺(tái)主機(jī)，并行操作可以節(jié)約大量的時(shí)間。

當(dāng)被操作對(duì)象數(shù)目不大時(shí)，可以直接利用 multiprocessing 中的 Process 動(dòng)態(tài)生成多個(gè)進(jìn)程，十幾個(gè)還好，但如果是上百個(gè)，上千個(gè)目標(biāo)，手動(dòng)的去限制進(jìn)程數(shù)量卻又太過繁瑣，此時(shí)可以發(fā)揮進(jìn)程池的功效。

Pool可以提供指定數(shù)量的進(jìn)程供用戶調(diào)用，當(dāng)有新的請(qǐng)求提交到 pool 中時(shí)，如果進(jìn)程池還沒有滿，那么就會(huì)創(chuàng)建一個(gè)新的進(jìn)程用來(lái)執(zhí)行該請(qǐng)求；但如果池中的進(jìn)程數(shù)已經(jīng)達(dá)到規(guī)定最大值，那么該請(qǐng)求就會(huì)等待，直到池中有進(jìn)程結(jié)束，才會(huì)創(chuàng)建新的進(jìn)程來(lái)它。

Python multiprocessing 模塊提供了 Pool() 函數(shù)，專門用來(lái)創(chuàng)建一個(gè)進(jìn)程池，該函數(shù)的語(yǔ)法格式如下：

multiprocessing.Pool( processes )

其中，processes 參數(shù)用于指定該進(jìn)程池中包含的進(jìn)程數(shù)。

如果進(jìn)程是 None，則默認(rèn)使用 os.cpu_count() 返回的數(shù)字（根據(jù)本地的 cpu 個(gè)數(shù)決定，processes 小于等于本地的 cpu 個(gè)數(shù)）。

請(qǐng)看下面的實(shí)例：

from multiprocessing import Pool
import os
import time
import random

def worker(msg):
  t_start = time.time()
  print("%s開始執(zhí)行,進(jìn)程號(hào)為%d" % (msg, os.getpid()))
  # random.random()隨機(jī)生成0~1之間的浮點(diǎn)數(shù)
  time.sleep(random.random()*2)
  t_stop = time.time()
  print(msg, "執(zhí)行完畢，耗時(shí)%0.2f" % (t_stop-t_start))

if __name__ == "__main__":
  po = Pool(3) # 定義一個(gè)進(jìn)程池，最大進(jìn)程數(shù)3
  for i in range(0, 8):
    # Pool().apply_async(要調(diào)用的目標(biāo),(傳遞給目標(biāo)的參數(shù)元祖,))
    # 每次循環(huán)將會(huì)用空閑出來(lái)的子進(jìn)程去調(diào)用目標(biāo)
    po.apply_async(worker, (i,))

  print("----start----")
  # 關(guān)閉進(jìn)程池，關(guān)閉后po不再接收新的請(qǐng)求
  po.close()
  # 等待po中所有子進(jìn)程執(zhí)行完成，必須放在close語(yǔ)句之后
  po.join()
  print("-----end-----")

運(yùn)行結(jié)果：

multiprocessing.Pool 常用方法說(shuō)明

apply_async(func[, args[, kwds]]) ：使用非阻塞方式調(diào)用 func（并行執(zhí)行，堵塞方式必須等待上一個(gè)進(jìn)程退出才能執(zhí)行下一個(gè)進(jìn)程），args 為傳遞給 func 的參數(shù)列表，kwds 為傳遞給 func 的關(guān)鍵字參數(shù)列表。

close()：關(guān)閉 Pool，使其不再接受新的任務(wù)。

terminate()：不管任務(wù)是否完成，立即終止。

join()：主進(jìn)程阻塞，等待子進(jìn)程的退出，必須在 close 或 terminate 之后使用。

進(jìn)程池中的 Queue

如果要使用 Pool 創(chuàng)建進(jìn)程，就需要使用 multiprocessing.Manager() 中的 Queue()，而不是 multiprocessing.Queue()，否則會(huì)得到一條如下的錯(cuò)誤信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

下面的實(shí)例演示了進(jìn)程池中的進(jìn)程如何通信：

from multiprocessing import Manager, Pool
import os
import time
import random

def writer(q):
  print("writer啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))
  for i in "xiaoming":
    q.put(i)

def reader(q):
  print("reader啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))
  for i in range(q.qsize()):
    print("reader從Queue獲取到消息：%s" % q.get(True))

if __name__ == "__main__":
  print("(%s) start" % os.getpid())
  # 使用Manager中的Queue
  q = Manager().Queue()
  po = Pool()
  po.apply_async(writer, (q,))
  # 先讓上面的任務(wù)向Queue存入數(shù)據(jù)，然后再讓下面的任務(wù)開始從中取數(shù)據(jù)
  time.sleep(1)
  po.apply_async(reader, (q,))
  po.close()
  po.join()
  print("(%s) End" % os.getpid())

運(yùn)行結(jié)果：

(17528) start
writer啟動(dòng)(2216),父進(jìn)程為(17528)
reader啟動(dòng)(2216),父進(jìn)程為(17528)
reader從Queue獲取到消息：x
reader從Queue獲取到消息：i
reader從Queue獲取到消息：a
reader從Queue獲取到消息：o
reader從Queue獲取到消息：m
reader從Queue獲取到消息：i
reader從Queue獲取到消息：n
reader從Queue獲取到消息：g
(17528) End

以上就是python 進(jìn)程池pool使用詳解的詳細(xì)內(nèi)容，更多關(guān)于python 進(jìn)程池pool的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: