Python 多处理：如何知道使用池或进程？

Question

所以我正在编写一个算法，函数multiprocess 应该在与cpus 一样多的进程上并行调用另一个函数CreateMatrixMp()。我以前从未做过多处理，并且无法确定以下哪种方法更有效。在函数CreateMatrixMp() 的上下文中使用的“高效”一词可能需要被调用数千次。我已经阅读了python multiprocessing 模块上的所有文档，并得出了这两种可能性：

首先是使用Pool类：

def MatrixHelper(self, args):
    return self.CreateMatrix(*args)

def Multiprocess(self, sigmaI, sigmaX):

    cpus = mp.cpu_count()
    print('Number of cpu\'s to process WM: %d' % cpus)
    poolCount = cpus*2
    args = [(sigmaI, sigmaX, i) for i in range(self.numPixels)]

    pool = mp.Pool(processes = poolCount, maxtasksperchild= 2)
    tempData = pool.map(self.MatrixHelper, args)
    pool.close()
    pool.join()

接下来是使用Process 类：

def Multiprocess(self, sigmaI, sigmaX):

    cpus = mp.cpu_count()
    print('Number of cpu\'s to process WM: %d' % cpus)

    processes = [mp.Process(target = self.CreateMatrixMp, args = (sigmaI, sigmaX, i,)) for i in range(self.numPixels)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

Pool 似乎是更好的选择。我读过它会减少开销。而Process不考虑机器上cpu的数量。唯一的问题是，以这种方式使用Pool 会给我一个又一个错误，每当我修复一个时，它下面就会出现一个新的。 Process 似乎更容易实现，据我所知，这可能是更好的选择。你的经历告诉你什么？

如果应该使用Pool，那么我选择map()对吗？最好维持秩序。我有tempData = pool.map(...)，因为map 函数应该返回每个进程的结果列表。我不确定Process 如何处理其返回的数据。

Answer 1

我认为Pool 类通常更方便，但这取决于您希望结果是有序的还是无序的。

假设您要创建 4 个随机字符串（例如，可能是随机用户 ID 生成器等）：

import multiprocessing as mp
import random
import string

# Define an output queue
output = mp.Queue()

# define a example function
def rand_string(length, output):
    """ Generates a random string of numbers, lower- and uppercase chars. """
    rand_str = ''.join(random.choice(
                    string.ascii_lowercase
                    + string.ascii_uppercase
                    + string.digits)
               for i in range(length))
    output.put(rand_str)

# Setup a list of processes that we want to run
processes = [mp.Process(target=rand_string, args=(5, output)) for x in range(4)]

# Run processes
for p in processes:
    p.start()

# Exit the completed processes
for p in processes:
    p.join()

# Get process results from the output queue
results = [output.get() for p in processes]

print(results)

# Output
# ['yzQfA', 'PQpqM', 'SHZYV', 'PSNkD']

在这里，顺序可能无关紧要。我不确定是否有更好的方法来做到这一点，但如果我想按照函数调用的顺序跟踪结果，我通常会返回带有 ID 作为第一项的元组，例如，

# define a example function
def rand_string(length, pos, output):
    """ Generates a random string of numbers, lower- and uppercase chars. """
    rand_str = ''.join(random.choice(
                    string.ascii_lowercase
                    + string.ascii_uppercase
                    + string.digits)
                for i in range(length))
    output.put((pos, rand_str))

# Setup a list of processes that we want to run
processes = [mp.Process(target=rand_string, args=(5, x, output)) for x in range(4)]

print(processes)

# Output
# [(1, '5lUya'), (3, 'QQvLr'), (0, 'KAQo6'), (2, 'nj6Q0')]

然后让我对结果进行排序：

results.sort()
results = [r[1] for r in results]
print(results)

# Output:
# ['KAQo6', '5lUya', 'nj6Q0', 'QQvLr']

池类

现在回答您的问题：这与 Pool 类有何不同？ 您通常更喜欢 Pool.map 返回有序的结果列表，而无需通过创建元组和按 ID 对它们进行排序的过程。因此，我会说它通常更有效。

def cube(x):
    return x**3

pool = mp.Pool(processes=4)
results = pool.map(cube, range(1,7))
print(results)

# output:
# [1, 8, 27, 64, 125, 216]

等价的还有一个“apply”方法：

pool = mp.Pool(processes=4)
results = [pool.apply(cube, args=(x,)) for x in range(1,7)]
print(results)

# output:
# [1, 8, 27, 64, 125, 216]

Pool.apply 和 Pool.map 都会锁定主程序，直到进程完成。

现在，您还有Pool.apply_async 和Pool.map_async，它们会在进程完成后立即返回结果，这与上面的Process 类基本相似。优点可能是它们为您提供了方便的 apply 和 map 功能，您可以从 Python 的内置 apply 和 map 中了解这些功能。

Answer 2

您可以使用pypeln 轻松做到这一点：

import pypeln as pl

stage = pl.process.map(
    CreateMatrixMp, 
    range(self.numPixels), 
    workers=poolCount, 
    maxsize=2,
)

# iterate over it in the main process
for x in stage:
   # code

# or convert it to a list
data = list(stage)