快捷導(dǎo)航

Python?多處理模塊如何使用示例詳解

更新時(shí)間：2023年09月04日 10:14:31 作者：冷凍工廠

這篇文章主要為大家介紹了Python?多處理模塊如何使用示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

什么是多處理模塊？

在本文中，我們將學(xué)習(xí)如何使用多處理模塊中的特定 Python 類（進(jìn)程類）。我將通過(guò)示例為您提供快速概述。

還有什么比從官方文檔中提取模塊更好的方式來(lái)描述模塊呢？ Multiprocessing 是一個(gè)使用類似于線程模塊的 API 支持生成進(jìn)程的包。多處理包提供本地和遠(yuǎn)程并發(fā)，通過(guò)使用子進(jìn)程而不是線程有效地回避全局解釋器鎖。

線程模塊不是本文的重點(diǎn)，但總而言之，線程模塊將處理一小段代碼執(zhí)行（輕量級(jí)且具有共享內(nèi)存），而多處理模塊將處理程序執(zhí)行（較重且完全隔離）。

一般來(lái)說(shuō)，多處理模塊提供了各種其他類、函數(shù)和實(shí)用程序，可用于處理程序執(zhí)行期間執(zhí)行的多個(gè)進(jìn)程。如果程序需要在其工作流程中應(yīng)用并行性，該模塊專門(mén)設(shè)計(jì)為交互的主要點(diǎn)。我們不會(huì)討論多處理模塊中的所有類和實(shí)用程序，而是將重點(diǎn)關(guān)注一個(gè)非常具體的類，即進(jìn)程類。

什么是進(jìn)程類？

在本節(jié)中，我們將嘗試更好地介紹進(jìn)程是什么，以及如何在 Python 中識(shí)別、使用和管理進(jìn)程。正如 GNU C 庫(kù)中所解釋的：“進(jìn)程是分配系統(tǒng)資源的基本單位。每個(gè)進(jìn)程都有自己的地址空間和（通常）一個(gè)控制線程。一個(gè)進(jìn)程執(zhí)行一個(gè)程序；可以讓多個(gè)進(jìn)程執(zhí)行相同的程序程序，但每個(gè)進(jìn)程在其自己的地址空間內(nèi)都有自己的程序副本，并獨(dú)立于其他副本執(zhí)行它。”

但這在 Python 中是什么樣子的呢？到目前為止，我們已經(jīng)設(shè)法對(duì)進(jìn)程是什么、進(jìn)程和線程之間的區(qū)別進(jìn)行了一些描述和參考，但到目前為止我們還沒(méi)有觸及任何代碼。好吧，讓我們改變一下，用 Python 做一個(gè)非常簡(jiǎn)單的流程示例：

#!/usr/bin/env python
import os
# A very, very simple process.
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")

這將產(chǎn)生以下輸出：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 144112

正如您所看到的，任何正在運(yùn)行的 Python 腳本或程序都是它自己的一個(gè)進(jìn)程。

創(chuàng)建子進(jìn)程

那么在父進(jìn)程中生成不同的子進(jìn)程又如何呢？好吧，要做到這一點(diǎn)，我們需要多處理模塊中的 Process 類的幫助，它看起來(lái)像這樣：

#!/usr/bin/env python
import os
import multiprocessing
def child_process():
    print(f"Hi! I'm a child process {os.getpid()}")
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    # Here we create a new instance of the Process class and assign our
    # `child_process` function to be executed.
    process = multiprocessing.Process(target=child_process)
    # We then start the process
    process.start()
    # And finally, we join the process. This will make our script to hang and
    # wait until the child process is done.
    process.join()

這將產(chǎn)生以下輸出：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 144078
Hi! I'm a child process 144079

關(guān)于上一個(gè)腳本的一個(gè)非常重要的注意事項(xiàng)：如果您不使用 process.join() 來(lái)等待子進(jìn)程執(zhí)行并完成，那么該點(diǎn)的任何其他后續(xù)代碼將實(shí)際執(zhí)行，并且可能會(huì)變得有點(diǎn)難以同步您的工作流程。

考慮以下示例：

#!/usr/bin/env python
import os
import multiprocessing
def child_process():
    print(f"Hi! I'm a child process {os.getpid()}")
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    # Here we create a new instance of the Process class and assign our
    # `child_process` function to be executed.
    process = multiprocessing.Process(target=child_process)
    # We then start the process
    process.start()
    # And finally, we join the process. This will make our script to hang and
    # wait until the child process is done.
    #process.join()
    print("AFTER CHILD EXECUTION! RIGHT?!")

該代碼片段將產(chǎn)生以下輸出：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 145489
AFTER CHILD EXECUTION! RIGHT?!
Hi! I'm a child process 145490

當(dāng)然，斷言上面的代碼片段是錯(cuò)誤的也是不正確的。這完全取決于您想要如何使用該模塊以及您的子進(jìn)程將如何執(zhí)行。所以要明智地使用它。

創(chuàng)建各種子進(jìn)程

如果要生成多個(gè)進(jìn)程，可以利用 for 循環(huán)（或任何其他類型的循環(huán)）。它們將允許您創(chuàng)建對(duì)所需流程的盡可能多的引用，并在稍后階段啟動(dòng)/加入它們。

#!/usr/bin/env python
import os
import multiprocessing
def child_process(id):
    print(f"Hi! I'm a child process {os.getpid()} with id#{id}")
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    list_of_processes = []
    # Loop through the number 0 to 10 and create processes for each one of
    # them.
    for i in range(0, 10):
        # Here we create a new instance of the Process class and assign our
        # `child_process` function to be executed. Note the difference now that
        # we are using the `args` parameter now, this means that we can pass
        # down parameters to the function being executed as a child process.
        process = multiprocessing.Process(target=child_process, args=(i,))
        list_of_processes.append(process)
    for process in list_of_processes:
        # We then start the process
        process.start()
        # And finally, we join the process. This will make our script to hang
        # and wait until the child process is done.
        process.join()

這將產(chǎn)生以下輸出：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 146056
Hi! I'm a child process 146057 with id#0
Hi! I'm a child process 146058 with id#1
Hi! I'm a child process 146059 with id#2
Hi! I'm a child process 146060 with id#3
Hi! I'm a child process 146061 with id#4
Hi! I'm a child process 146062 with id#5
Hi! I'm a child process 146063 with id#6
Hi! I'm a child process 146064 with id#7
Hi! I'm a child process 146065 with id#8
Hi! I'm a child process 146066 with id#9

數(shù)據(jù)通信

在上一節(jié)中，我描述了向 multiprocessing.Process 類構(gòu)造函數(shù)添加一個(gè)新參數(shù) args。此參數(shù)允許您將值傳遞給子進(jìn)程以在函數(shù)內(nèi)部使用。但你知道如何從子進(jìn)程返回?cái)?shù)據(jù)嗎？

您可能會(huì)認(rèn)為，要從子級(jí)返回?cái)?shù)據(jù)，必須使用其中的 return 語(yǔ)句才能真正檢索數(shù)據(jù)。進(jìn)程非常適合以隔離的方式執(zhí)行函數(shù)，而不會(huì)干擾共享資源，這意味著我們知道從函數(shù)返回?cái)?shù)據(jù)的正常且常用的方式。在這里，由于其隔離而不允許。

相反，我們可以使用隊(duì)列類，它將為我們提供一個(gè)在父進(jìn)程與其子進(jìn)程之間通信數(shù)據(jù)的接口。在這種情況下，隊(duì)列是一個(gè)普通的 FIFO（先進(jìn)先出），具有用于處理多處理的內(nèi)置機(jī)制。

考慮以下示例：

#!/usr/bin/env python
import os
import multiprocessing
def child_process(queue, number1, number2):
    print(f"Hi! I'm a child process {os.getpid()}. I do calculations.")
    sum = number1 + number2
    # Putting data into the queue
    queue.put(sum)
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    # Defining a new Queue()
    queue = multiprocessing.Queue()
    # Here we create a new instance of the Process class and assign our
    # `child_process` function to be executed. Note the difference now that
    # we are using the `args` parameter now, this means that we can pass
    # down parameters to the function being executed as a child process.
    process = multiprocessing.Process(target=child_process, args=(queue,1, 2))
    # We then start the process
    process.start()
    # And finally, we join the process. This will make our script to hang and
    # wait until the child process is done.
    process.join()
    # Accessing the result from the queue.
    print(f"Got the result from child process as {queue.get()}")

它將給出以下輸出：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 149002
Hi! I'm a child process 149003. I do calculations.
Got the result from child process as 3

異常處理

處理異常是一項(xiàng)特殊且有些困難的任務(wù)，我們?cè)谑褂昧鞒棠K時(shí)必須不時(shí)地完成它。原因是，默認(rèn)情況下，子進(jìn)程內(nèi)發(fā)生的任何異常將始終由生成它的 Process 類處理。

下面的代碼引發(fā)帶有文本的異常：

#!/usr/bin/env python
import os
import multiprocessing
def child_process():
    print(f"Hi! I'm a child process {os.getpid()}.")
    raise Exception("Oh no! :(")
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    # Here we create a new instance of the Process class and assign our
    # `child_process` function to be executed. Note the difference now that
    # we are using the `args` parameter now, this means that we can pass
    # down parameters to the function being executed as a child process.
    process = multiprocessing.Process(target=child_process)
    try:
        # We then start the process
        process.start()
        # And finally, we join the process. This will make our script to hang and
        # wait until the child process is done.
        process.join()
        print("AFTER CHILD EXECUTION! RIGHT?!")
    except Exception:
        print("Uhhh... It failed?")

輸出結(jié)果：

[r0x0d@fedora ~]$ python /tmp/tmp.iuW2VAurGG/scratch.py
Hi! I'm process 149505
Hi! I'm a child process 149506.
Process Process-1:
Traceback (most recent call last):
File "/usr/lib64/python3.11/multiprocessing/process.py", line 314, in _bootstrap
self.run()
File "/usr/lib64/python3.11/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/tmp/tmp.iuW2VAurGG/scratch.py", line 7, in child_process
raise Exception("Oh no! :(")
Exception: Oh no! :(
AFTER CHILD EXECUTION! RIGHT?!

如果您跟蹤代碼，您將能夠注意到在 process.join() 調(diào)用之后仔細(xì)放置了一條 print 語(yǔ)句，以模擬父進(jìn)程仍在運(yùn)行，即使在子進(jìn)程中引發(fā)了未處理的異常之后也是如此。

克服這種情況的一種方法是在子進(jìn)程中實(shí)際處理異常，如下所示：

#!/usr/bin/env python
import os
import multiprocessing
def child_process():
    try:
        print(f"Hi! I'm a child process {os.getpid()}.")
        raise Exception("Oh no! :(")
    except Exception:
        print("Uh, I think it's fine now...")
if __name__ == "__main__":
    print(f"Hi! I'm process {os.getpid()}")
    # Here we create a new instance of the Process class and assign our
    # `child_process` function to be executed. Note the difference now that
    # we are using the `args` parameter now, this means that we can pass
    # down parameters to the function being executed as a child process.
    process = multiprocessing.Process(target=child_process)
    # We then start the process
    process.start()
    # And finally, we join the process. This will make our script to hang and
    # wait until the child process is done.
    process.join()
    print("AFTER CHILD EXECUTION! RIGHT?!")

現(xiàn)在，您的異常將在您的子進(jìn)程內(nèi)處理，這意味著您可以控制它會(huì)發(fā)生什么以及在這種情況下應(yīng)該做什么。