PYTHON 中的多进程

落日映苍穹つ 2023-10-03 08:39 15阅读 0赞

*随着耗电应用程序数量的增加，在某些情况下对速度和低延迟的需求已成为挑战。但是，具有多个处理器/多核处理器的机器的可用性帮助我们应对这种情况。这篇文章将指导您在 python 中使用多处理。*

### **介绍** ###

在当代，许多 CPU 都采用多核制造，以通过实现应用程序的并行性和并发性来提高性能。因此，可以利用此类 CPU 来编写可以并行运行的代码，从而加快应用程序的速度。这就是 Python Multiprocessing 包发挥作用的地方。这个 API 与 python 多线程模块非常相似。

上一篇关于[Python 中的多线程的][Python]帖子提供了关于 python 线程模块的清晰解释，如果您还没有，[请单击此处][Python]通读相同内容。

### **目录** ###

*  什么是 Python 中的多处理
 *  Python 多处理示例
 *  多处理用例
 *  多处理与多线程
 *  多处理池
 *  并行处理与串行处理性能测试
 *  在进程之间共享数据 - 共享内存
 *  多处理队列
 *  多处理值和锁
 *  多处理管理器
 *  结论

### **什么是 Python 中的多处理？** ###

简单来说，多处理或并行处理被定义为单个中央计算系统范围内的应用程序使用两个或多个处理器。Python 多处理模块提供了一个干净且直观的 API 来利用 Python 中的并行处理。

所有进程彼此独立，并拥有自己的计算资源（例如内存和处理能力）。

要访问进程之间的对象，必须使用进程间通信 (IPC) 的概念，这也将在本文中讨论。

### **Python 多处理示例** ###

让我们从 Python 中的一个简单多处理示例开始，将一组数字的平方和平方根计算为 2 个不同的进程。

我们将通过导入 Process 类创建一个 Process 对象并启动这两个进程。

下面是创建进程对象的语法

`from multiprocessing import Process`  
`Process(target=function_name, name='Process 1', args=(arg1, arg2, arg3), kwargs={arg4: 'abc'})`

args：接受元组参数列表  
kwargs：接受 dict 对象  
名称：字符串

from multiprocessing import Process
    import math
    
    def calc_square(numbers):
        """ Calculate the squares of the input set of numbers """
        squares = [i*i for i in numbers]
        print("Squares: {0}".format(squares))
            
    def calc_square_root(numbers):
        """ Calculate the square roots of the input set of numbers """
        square_roots = [round(math.sqrt(i), 2) for i in numbers]
        print("Square Roots: {0}".format(square_roots))
            
    if __name__ == "__main__":
        number_set = list(range(1,11))
        
        # Create the Processes
        p1 = Process(target=calc_square, args=(number_set,))
        p2 = Process(target=calc_square_root, args=(number_set,))
    
        # Start the Processes
        p1.start()
        p2.start()
    
    # Note: (1) is considered as int, whereas (1,) is considered as a tuple in python. So, in case 
    # of a single argument an extra comma must be placed to project it as a tuple

Squares: [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
    Square Roots: [1.0, 1.41, 1.73, 2.0, 2.24, 2.45, 2.65, 2.83, 3.0, 3.16]

### **多处理用例** ###

*  多处理最适合重量级任务。
 *  当您有多个可以独立运行的任务时。
 *  当任务之间不涉及大量数据共享时。虽然数据共享是可能的，但与线程相比使用起来相当复杂。
 *  当您的应用程序受 CPU 限制而不是 I/O 限制时。

### **多处理与多线程** ###

通常很多人对多处理和多线程的概念感到困惑。

以下是一些主要区别，

*  多处理启动彼此独立的不同进程。另一方面，多线程启动仍然依赖于父进程的线程。
 *  Multiprocessing 中的每个进程都有自己的 CPU 功率和内存，每个进程都不同。而在多线程中，各个线程使用父进程中存在的相同资源（CPU + 内存）。
 *  如果任何进程由于异常而失败，则其他进程在 Multiprocessing 的情况下仍将运行。相反，如果任何线程失败，则所有其他线程/任务都将终止。
 *  多处理需要一些特殊的对象或某种共享内存来访问不同进程中的对象。与此相反，线程之间共享对象/数据要容易得多，因为它们共享相同的内存空间。
 *  多处理更适合 CPU 密集型应用程序，而多线程最适合 I/O 密集型应用程序。

![图像.png][.png]

有关 Python 中多线程的更多详细信息，[请单击此处][Python]。

### **多处理池** ###

Multiprocessing Pool 对象是这个包提供的最好的特性之一。与线程模块相比，这是该模块脱颖而出的地方。

#### **游泳池的特点：** ####

*  池将数据映射到相应的工作人员并将其聚合回来以显示最终结果。
 *  Pool 对象使您可以自由指定执行所需的工作进程数。例如，如果您指定 4 个 worker 并且您有 8 个作业要完成，其中 4 个将首先启动，当一个 worker 完成其作业时，它会立即开始执行下一个进程。
 *  如果未指定工作进程数，默认情况下 Pool 将使用您机器中存在的 CPU 内核数。不建议这样做，因为它可能会占用您机器中的所有资源，从而导致其他作业变慢。

下面的示例演示了 Pool 的用法。为了更好地说明进程的启动方式，使用`os.getpid`方法显示了各个进程的 process\_id 。

如下图，由于我们指定了4个worker，进程9990到9993会被递归使用，直到执行完成。

from multiprocessing import Pool
    import os
    
    def compute_squares(num):
        print(f"Process {os.getpid()} calculating square of {num}")
        return num*num
    
    if __name__ == "__main__":
        # Create the Pool object with the number of workers, 4 in this case.
        p = Pool(4)
        
        # Map the jobs to the individual workers
        result = p.map(compute_squares, range(12))
        
        # Close the Pool object
        p.close()
        print(f"Result is: {result}")

Process 9991 calculating square of 1
    Process 9992 calculating square of 2
    Process 9990 calculating square of 0
    Process 9993 calculating square of 3
    Process 9990 calculating square of 5
    Process 9992 calculating square of 4
    Process 9991 calculating square of 6
    Process 9992 calculating square of 7
    Process 9990 calculating square of 8
    Process 9991 calculating square of 9
    Process 9993 calculating square of 10
    Process 9990 calculating square of 11
    Result is: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121]

***提示：**实例化和调用 Pool 方法也可以以更简洁的方式完成。*

p = Pool(4)
    result = p.map(calc_squares, range(12))
    p.close()

以上3个语句可以替换为，

with Pool(4) as p:
        result = p.map(calc_squares, range(12)

#### **如果您的函数中有 1 个以上的参数怎么办？** ####

在这种情况下，您可以使用`pool.starmap`函数 (Python 3.3+) 或通过变通方法使用替代方法来发送 2 个参数。

下面是在 map 中使用 1 个以上参数的示例。可以将元组列表传递给中间函数，该函数进一步将这些元组解包为原始函数的 args。

from multiprocessing import Pool
    import os
    
    def calc_product(num1, num2, num3):
        print(f"Process {os.getpid()} calculating product of {num1, num2, num3}")
        return num1*num2*num3
    
    def exec_func(nums):
        """ Intermediate method to call calc_product method """
        return calc_product(*nums)
    
    if __name__ == "__main__":
        with Pool(4) as p:
            result = p.map(exec_func, [(1,2,3),(3,4,5)])
        print(f"Result is: {result}")

Process 11123 calculating product of (1, 2, 3)
    Process 11124 calculating product of (3, 4, 5)
    Result is: [6, 60]

#### **池星图** ####

星图可从 Python 3.3 获得。传递给星图的元素必须是由可迭代对象组成的可迭代对象。如果您使用的是旧版本的 Python，则可以使用以前的方法。

from multiprocessing import Pool
    import os
    
    def calc_product(num1, num2, num3):
        print(f"Process {os.getpid()} calculating product of {num1, num2, num3}")
        return num1*num2*num3
    
    if __name__ == "__main__":
        with Pool(2) as p:
            result = p.starmap(calc_product, [(1,2,3),(3,4,5)])
        print(f"Result is: {result}")

Process 11144 calculating product of (1, 2, 3)
    Process 11145 calculating product of (3, 4, 5)
    Result is: [6, 60]

### **并行处理与串行处理性能测试** ###

让我们使用Pool方法在python中进行串行处理和并行处理之间的简单基准测试

from multiprocessing import Pool
    import time
    
    def calc_cubes(_):
        """ Calculate the cube of numbers ranging till 1000 """
        return list(map(lambda n: n*n*n, range(1000)))
    
    if __name__ == "__main__":
        t1 = time.time()
        with Pool(5) as p:
            p.map(calc_cubes, range(100000))  # Call the calc_cubes method 100000 times with 5 concurrent tasks.
        print("Parallel Processing Time Taken: {0}".format(time.time() - t1))
        
        print("Starting Serial Processing")
        t2 = time.time()
        list(map(calc_cubes, range(100000)))  # Call the calc_cubes method 100000 times sequentially.
        print("Sequential Processing Time Taken: {0}".format(time.time() - t2))

Parallel Processing Time Taken: 12.06329607963562
    Starting Serial Processing
    Sequential Processing Time Taken: 20.98725724220276

从上面的例子中，我们可以看到，当通过单个进程完成时，我们的计算时间几乎翻了一番。

### **在进程之间共享数据 - 共享内存** ###

在了解如何在进程之间共享数据之前，让我们先熟悉一下不同进程的基础知识。

每个进程都有自己的资源份额，这意味着内存和 CPU 分配是每个进程本机的。

每个进程都将拥有自己的内存或地址空间，这些空间或地址空间仅适用于该进程。因此，这些过程中的数据将存储在这些单独的内存中。因此，共享对象需要进程之间的某种通信。

![图像.png][.png 1]

为了在这些进程之间共享数据，进程之间必须存在某种通信。这就是*进程间通信 (IPC)*概念发挥作用的地方。我们需要一个公共存储区或共享内存来从不同的启动进程访问这些对象。

Multiprocessing Module 提供了一些专门设计用于在进程之间共享数据的对象。下面描述了其中一些方法。

**注意：**如果您的进程不需要很多 CPU Power 但涉及很多常见的共享变量，那么在这种情况下建议使用线程模块。

### **多处理队列** ###

MultiProcessing Queue 类似于 Python 中的 Queue 类

下面是从输入数字集中获取偶数并将其放入多处理队列的示例。

from multiprocessing import Queue, Process
    import time
    
    def get_even(nums, q):
        """ Obtain the even numbers from a set of input numbers Multiprocessing Queue is passed as an additional argument Queue will consist of all even numbers """
        q.put([n for n in nums if n%2 == 0])
    # q.put(list(filter(lambda n: n%2==0, nums))) # alternate method
    
    if __name__ == "__main__":
        q = Queue()
        p = Process(target=get_even, args=(range(100), q))
        p.start()
        p.join()
        
        # Display the contents of the queue until it is empty
        while not q.empty():
            print(f"Queue: {q.get()}")

Queue: [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98]

**注意：** Join 必须用于您希望程序在执行 join 语句之后出现的代码之前等待的情况。

例如：  
create process  
process.join()  
print(“complete”) – 只有在进程完成后才会打印。

### **多处理值和锁** ###

如果您需要维护和修改进程之间的共享变量，我们可以使用模块中的 Value 对象。

创建 Value 对象的语法是，

`val = Value(typecode, *args)`

常用的类型代码是，  
'i' - Integer  
'd' - Float  
'u' - Unicode String

让我们模拟一个例子，我们将一个数字乘以和除以 10，最终得到结果 1。

#### **锁定前：** ####

from multiprocessing import Process, Value
    import time
    
    def add(val):
        for i in range(10000):
            val.value += 1
            
    def subtract(val):
        for i in range(10000):
            val.value -= 1
            
    if __name__ == "__main__":
        
        # Create Value Object
        val = Value('i', 1) # i represents integer and 1 is the initial value
        
        p1 = Process(target=add, args=(val, ))
        p2 = Process(target=subtract, args=(val, ))
        
        # Start the Tasks
        p1.start()
        p2.start()
        
        # Wait for the Tasks to complete
        p1.join()
        p2.join()
        
        print("Value is {0}".format(val.value))

Value is 273

如上所示，实际得到的结果与预期的结果 1 不同。这是因为两个进程都在访问同一个对象，导致两个进程同时修改值发生冲突。

让我们使用 Multiprocessing Lock 运行相同的示例。

#### **锁定后：** ####

from multiprocessing import Process, Value, Lock
    import time
    
    def add(val, lock):
        for i in range(10000):
            lock.acquire() # Acquire Lock
            val.value += 1
            lock.release() # Release the lock
            
    def subtract(val, lock):
        for i in range(10000):
            with lock:     # Alternate method to create a lock
                val.value -= 1
            
    if __name__ == "__main__":
        lock = Lock()
        val = Value('i', 1) # i represents integer and 1 is the initial value
        p1 = Process(target=add, args=(val, lock))
        p2 = Process(target=subtract, args=(val, lock))
        
        p1.start()
        p2.start()
        
        p1.join()
        p2.join()
        
        print("Value is {0}".format(val.value))

Value in add: 976
    Value in subtract: 1
    Value is 1

### **多处理管理器** ###

多处理管理器维护一个独立的服务器进程，其中保存这些 python 对象。

Multiprocessing Library 还提供了 Manager 类，它可以访问更多的同步对象以在进程之间使用。

Manager 对象支持列表、字典、数组、队列、值等类型。

#### **何时使用多处理管理器？** ####

*  当需要使用列表、字典等同步对象时。
 *  如果进程在网络上运行并需要数据共享，那么多处理管理器是一个很好的选择。

下面的示例显示了如何在工作池中使用管理器列表

from multiprocessing import Pool, Manager
    import sys
    
    def update_string(val, l):
        l.append(val)
    
    if __name__ == "__main__":
        strs = ['with', 'great', 'power', 'comes', 'great', 'responsibilty']
        
        # Create the Manager Object
        mgr = Manager()
        
        # Create a manager list
        li = mgr.list()
        
        with Pool(4) as p:
            p.starmap(update_string, list(map(lambda x: (x,li), strs)))
        print(f"List: {li}")

List: ['with', 'great', 'power', 'comes', 'responsibilty', 'great']

下面的示例显示了如何使用管理器字典

import multiprocessing
    
    
    def worker(d, key, value):
        d[key] = value
    
    
    if __name__ == '__main__':
        mgr = multiprocessing.Manager()
        d = mgr.dict()
        jobs = [multiprocessing.Process(target=worker, args=(d, i, i * 2))
                for i in range(10)
                ]
        for j in jobs:
            j.start()
        for j in jobs:
            j.join()
        print('Results:', d)

结果：\{0: 0, 1: 2, 2: 4, 3: 6, 4: 8, 5: 10, 6: 12, 7: 14, 8: 16, 9: 18\}

###  ###

### **结论** ###

到目前为止，您必须熟悉python中多处理的概念，并且也清楚多处理和多线程之间的区别。

现在是使用并行处理加速应用程序的时候了。但请记住，强大的力量伴随着巨大的责任。

[Python]: https://datanoon.com/blog/multithreading_in_python
[.png]: https://img-blog.csdnimg.cn/img_convert/0167938e8d02c5d94d69b14cca598703.png
[.png 1]: https://img-blog.csdnimg.cn/img_convert/829993331020851fb77c900e90a2bc95.png