Python 多处理拒绝拆分我的可迭代对象

Question

我正在尝试编写一个多处理程序，似乎我已经完成了，并且我已经使用 System Monitor 应用程序验证了 Python 进程已创建。但问题是，几乎所有这些似乎都没有在现实中使用。在我的程序中，我试图将音频文件分割成块，所以我不认为它是我在其他线程中读到的“微不足道的计算负载”。

对我来说显示相同行为的最小示例：

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

def myfunc(myli):
    print(len(myli))
    for item in myli:
        x = item*item*item
        time.sleep(2)
    return

mylist = [random.randint(1,10000) for _ in range (0, 19999)]

with Pool(processes=8) as p, tqdm(total=len(mylist)) as pbar:
    for _ in p.imap_unordered(func=myfunc, iterable=(mylist,)):
        pbar.update()

如您所见，我在使用的func 内添加了一个print()，并且每次打印整个数组的长度。好像没有发生分裂一样。

我天真地尝试过使用不同的块大小并删除 tqdm（好像它起了任何作用）。

如果您能给我任何见解，我将不胜感激。

Answer 1

代码正在做你告诉它做的事情：你传递了一个长度为 1 的迭代，一个包含单个项目的元组 (mylist)。所以它将单个项目传递给单个工作人员进行处理。

但您不能改为使用iterable=mylist，因为myfunc() 期望得到一个序列，而不是整数。无论可迭代对象是什么，multiprocessing 一次将一个元素传递给工作人员。 chunksize 与此无关。无论chunksize 是1 还是10 亿，worker 函数一次只能看到一个元素。 chunksize 是一种隐蔽的优化，纯粹是为了减少所需的昂贵的进程间通信调用的数量。

如果您想将序列拆分为块并使用期望块的工作函数，那么您必须自己进行“分块”。例如，添加

# Generate slices of `xs` of length (at most) `n`.
def chunkit(xs, n):
    start = 0
    while start < len(xs):
        yield xs[start : start + n]
        start += n

并传递iterable=chunkit(mylist, 40)。然后所有 8 个进程都会很忙。一个将在mylist[0:40] 上工作，另一个在mylist[40:80] 上工作，另一个在mylist[80:120] 上工作，以此类推，直到mylist 用尽。