python - 具有常量和可迭代参数的函数的多重处理

Question

您好 stackoverflow 用户，

我试图查找这一点，但找不到答案:我本质上喜欢并行处理一个函数(独立进程!)并且该函数有一个可迭代(x)和几个常量参数(k、d)。这是一个非常简单的示例:

from multiprocessing import *

def test_function(args):
    k = args[0]
    d = args[1]
    x = args[2]
    del args

    return k*x + d

if __name__ == '__main__':
    pool = Pool(processes=2)

    k = 3.
    d = 5.

    constants = [k,d]
    xvalues = range(0,10)
    result = [pool.apply_async(test_function, constants.append(i)) for i in xvalues]

    output = [r.get() for r in result]

    print output
    #I expect [5.0, 8.0, 11.0, 14.0, 17.0, 20.0, 23.0, 26.0, 29.0, 32.0]

这给了我以下错误消息:

Traceback (most recent call last):
  File "test_function.py", line 23, in <module>
    output = [r.get() for r in result]
  File "C:\Program Files\Python2.7\lib\multiprocessing\pool.py", line 528, in get
    raise self._value
TypeError: test_function() argument after * must be a sequence, not NoneType

所以我的问题是:

此错误消息的实际含义是什么？

如何修复它以获得预期结果(请参阅最后一行代码示例)？

调用 apply_sync 的线路有更好/有效/优雅的方式吗？

仅供引用:我是 Python 新手，请耐心等待，如果我的帖子需要更多详细信息，请告诉我。

非常感谢您的建议!

Answer 1

What does this error message actually mean?

append 方法返回的值始终为 None，因此执行时:

pool.apply_async(test_function, constants.append(i))

您正在使用 None 作为 args 参数调用 pool.apply_asynch，但 apply_asynch 需要一个 可迭代作为参数。 apply_asynch 所做的事情称为 tuple-unpacking .

How do I fix it to get the expected results?

要实现预期的输出，只需将 i 连接到常量即可:

pool.apply_asynch(test_function, (constants + [i],))

Is there a better/working/elegant way for the line that calls apply_sync?

请注意，您必须将所有参数包装到一个元素元组中，因为您的 test_function 接受单个参数。您可以这样修改它:

def test_function(k, d, x):
    # etc

只需使用:

pool.apply_asynch(test_function, constants + [i])

apply_asynch将使用 tuple 解包自动将 args 解包到函数的三个参数中。 (仔细阅读 Pool.apply 和 friend 的文档)。

<小时/>

Is there a better/working/elegant way for the line that calls apply_sync?

正如 Silas 所指出的，您应该使用 Pool.map 而不是对值列表使用 Pool.apply_asynch 。或 Pool.map_asynch 方法，它们可以为您完成此操作。

例如:

results = pool.map(test_function, [(constants + [i],) for i in xvalues])

但是请注意，在这种情况下 test_function 必须接受单个参数，因此您必须手动解压常量和 x，就像您在问题中所做的那样。

<小时/>

此外，作为一般建议:

• 在您的test_function中，绝对不需要执行del args。它只会减慢函数的执行速度(稍微减慢)。仅在需要时谨慎使用 del。

• 您可以使用以下语法，而不是手动分配元组中的元素:

  k, d, x = args
  

  这相当于(可能稍慢):

  k = args[0]
  d = args[1]
  x = args[2]
  

• 使用多处理来调用如此简单的函数，预计会减慢。通信和同步进程的成本相当大，因此您必须避免调用简单的函数，并尽可能尝试“分块”工作(例如，不要单独发送每个请求，而是将 100 个请求的列表发送给一个工作人员)论证)。