1

嗨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 新手,请多多包涵,如果我的帖子需要更多详细信息,请告诉我。

非常感谢您的任何建议!

4

2 回答 2

1

此错误消息实际上是什么意思?

append方法返回的值总是None,因此在执行以下操作时:

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

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

如何修复它以获得预期的结果?

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

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

对于调用 apply_sync 的线路是否有更好/工作/优雅的方式?

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

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

并简单地使用:

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

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


对于调用 apply_sync 的线路是否有更好/工作/优雅的方式?

正如 Silas 所指出的,Pool.apply_asynch您应该使用Pool.maporPool.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]
    
  • 使用调用这些简单的函数预计会大大降低速度。multiprocessing通信和同步进程的成本非常高,因此您必须避免调用简单的函数,并且尽可能尝试“分块”工作(例如,不要单独发送每个请求,而是一次向工作人员发送 100 个请求的列表争论)。
于 2013-09-16T18:01:42.913 回答
0

constants.append(i) 返回 None ,您应该先附加值,然后constants用作第二个参数。

>>> constants = []
>>> i = 2
>>> bug_value = constants.append(i)
>>> constants
[2]
>>> bug_value is None
True
>>> 

result = [pool.apply_async(test_function, constants+ [i]) for i in xvalues]确实使用

list + list 附加两个列表并返回结果列表。

于 2013-09-16T18:01:39.060 回答