快捷導(dǎo)航

Python進程間通訊與進程池超詳細講解

更新時間：2022年12月23日 09:28:46 作者：alwaysrun

進程彼此之間互相隔離，要實現(xiàn)進程間通信（IPC），multiprocessing模塊主要通過隊列方式，隊列：隊列類似于一條管道，元素先進先出，需要注意的一點是：隊列都是在內(nèi)存中操作,進程退出,隊列清空,另外,隊列也是一個阻塞的形態(tài)

進程間通訊

multiprocessing中提供了Pipe（一對一）和Queue（多對多）用于進程間通訊。

隊列Queue

隊列是一個可用于進程間共享的Queue（內(nèi)部使用pipe與鎖），其接口與普通隊列類似：

put(obj[, block[, timeout]])：插入數(shù)據(jù)到隊列（默認阻塞，且沒有超時時間）；

若設(shè)定了超時且隊列已滿，會拋出queue.Full異常；
隊列已關(guān)閉時，拋出ValueError異常

get([block[, timeout]])：讀取并刪除一個元素；

若設(shè)定了超時且隊列為空，會拋出queue.Empty異常；
隊列已關(guān)閉時，拋出ValueError異常；若已阻塞后，再關(guān)閉則會一直阻塞；

qsize()：返回一個近似隊列長度（因多進程原因，長度會有誤差）；

empty()/full()：隊列空或慢（因多進程原因，會有誤差）；

close()：關(guān)閉隊列；

當主進程（創(chuàng)建Queue的）關(guān)閉隊列時，子進程中的隊列并沒有關(guān)閉，所以getElement進程會一直阻塞等待（為保證能正常退出，需要設(shè)為后臺進程）：

def putElement(name, qu: multiprocessing.Queue):
    try:
        for i in range(10):
            qu.put(f"{name}-{i + 1}")
            time.sleep(.1)
    except ValueError:
        print("queue closed")
    print(f"{name}: put complete")
def getElement(name, qu: multiprocessing.Queue):
    try:
        while True:
            r = qu.get()
            print(f"{name} recv: {r}")
    except ValueError:
        print("queue closed")
    print(f"{name}: get complete")
if __name__ == '__main__':
    qu = multiprocessing.Queue(100)
    puts = [multiprocessing.Process(target=putElement, args=(f"send{i}", qu)) for i in range(10)]
    gets = [multiprocessing.Process(target=getElement, args=(f"recv{i}", qu), daemon=True) for i in range(2)]
    list(map(lambda f: f.start(), puts))
    list(map(lambda f: f.start(), gets))
    for f in puts:
        f.join()
    print("To close")
    qu.close() # 只是main中的close了，其他進程中的并沒有

管道Pipe

multiprocessing.Pipe([duplex])返回一個連接對象對(conn1, conn2)。若duplex為True（默認），創(chuàng)建的是雙向管道；否則conn1只能用于接收消息，conn2只能用于發(fā)送消息：

send()：發(fā)送消息；
recv()：接收消息；

進程間的Pipe基于fork機制建立：

主進程創(chuàng)建Pipe：Pipe的兩個Connections連接的的都是主進程；
創(chuàng)建子進程后，Pipe也被拷貝了一份：此時有了4個Connections；
主進程關(guān)閉一個Out Connection，子進程關(guān)閉一個In Connection：就建立好了一個輸入在主進程，輸出在子進程的管道。

def pipeProc(pipe):
    outPipe, inPipe = pipe
    inPipe.close() # 必須關(guān)閉，否則結(jié)束時不會收到EOFError異常
    try:
        while True:
            r = outPipe.recv()
            print("Recv:", r)
    except EOFError:
        print("RECV end")
if __name__ == '__main__':
    outPipe, inPipe = multiprocessing.Pipe()
    sub = multiprocessing.Process(target=pipeProc, args=((outPipe, inPipe),))
    sub.start()
    outPipe.close() # 必須在進程成功運行后，才可關(guān)閉
    with inPipe:
        for x in range(10):
            inPipe.send(x)
            time.sleep(.1)
    print("send complete")
    sub.join()

進程池Pool

雖然使用多進程能提高效率，但進程的創(chuàng)建與銷毀會消耗較長時間；同時，過多進程會引起頻繁的調(diào)度，也增加了開銷。

進程池中有固定數(shù)量的進程：

請求到來時，從池中取出一個進程來處理任務(wù)；理完畢后，進程并不立即關(guān)閉，而是再放回進程池中；
當池中進程數(shù)量不夠，請求就要等待，直到拿到空閑進程后才能繼續(xù)執(zhí)行；
池中進程的數(shù)量是固定的，隱藏同一時間最多有固定數(shù)量的進程在運行。

multiprocessing.Pool([processes[, initializer[, initargs]]])

processes：要創(chuàng)建進程數(shù)量（默認os.cpu_count()個），在需要時才會創(chuàng)建；
initializer(*initargs)：每個工作進程啟動時執(zhí)行的方法（一般processes為幾就執(zhí)行幾次）；

Pool類中主要方法：

apply(func[, args[, kwds]])：以阻塞方式，從池中獲取進程并執(zhí)行func(*args,**kwargs)；
apply_async(func[, args[, kwds[, callback[, error_callback]]]])：異步方式（從池中獲取一個進程）執(zhí)行func(*args,**kwargs)，返回AsyncResult；
map(func, iterable[, chunksize])/map_async：map的并行版本（可同時處理多個任務(wù)），異步時返回MapResult；
starmap(func, iterable[, chunksize])/starmap_async：與map的區(qū)別是允許傳入多個參數(shù)；
imap(func, iterable[, chunksize])：map的惰性版本（返回結(jié)果是可迭代對象），內(nèi)存消耗會低些，返回迭代器IMapIterator；
imap_unordered(func, iterable[, chunksize])：imap返回的結(jié)果順序與map順序是相同的，而此方法返回的順序是亂序的（不依次等待每個任務(wù)完成，先完成的先返回），返回迭代器IMapIterator；
close()：關(guān)閉，禁止繼續(xù)提交任務(wù)（已提交任務(wù)會繼續(xù)執(zhí)行完成）；
terminate()：立即終止所有任務(wù)；
join()：等待工作進程完成（必須已close或terminate了）；

def poolWorker():
    print(f"worker in process {os.getpid()}")
    time.sleep(1)
def poolWorkerOne(name):
    print(f"worker one {name} in process {os.getpid()}")
    time.sleep(random.random())
    return name
def poolWorkerTwo(first, second):
    res = first + second
    print(f"worker two {res} in process {os.getpid()}")
    time.sleep(1./(first+1))
    return res
def poolInit():
    print("pool init")
if __name__ == '__main__':
    workers = multiprocessing.Pool(5, poolInit) # poolInit會被調(diào)用5次（線程啟動時）
    with workers:
        for i in range(5):
            workers.apply_async(poolWorker)
        arg = [(i, i) for i in range(10)]
        workers.map_async(poolWorkerOne, arg)
        results = workers.starmap_async(poolWorkerTwo, arg) # 每個元素（元組）會被拆分為獨立的參數(shù)
        print("Starmap:", results.get())
        results = workers.imap_unordered(poolWorkerOne, arg)
        for r in results: # r是亂序的（若使用imap，則與輸入arg的順序相同）
            print("Unordered:", r)
    # 必須保證workers已close了
    workers.join()

到此這篇關(guān)于Python進程間通訊與進程池超詳細講解的文章就介紹到這了,更多相關(guān)Python進程間通訊內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: