Python 的 map_async 如何保持结果有序？

Question

我正在尝试探索 Python 的 py3.3 多处理库，但我注意到 map_async 函数中出现了一个我无法解释的奇怪结果。我一直期望从回调中存储的结果是“乱序的”。也就是说，如果我向工作进程提供许多任务，一些任务应该在其他任务之前完成，不一定按照它们被输入或存在于输入列表中的相同顺序。但是，我得到了一组与输入任务完全对应的有序结果。即使在故意尝试通过减慢某些进程来“破坏”某些进程（这可能会让其他进程在它之前完成）之后也是如此。

我在calculate 函数中有一个打印语句，显示它们被无序计算，但结果 仍然是有序的。虽然我不确定我是否可以相信打印语句是一个很好的指标，表明事情实际上是在计算乱序的。

测试过程（一般示例）： 构建一个对象列表，每个对象都包含一个整数。 将该对象列表作为参数提交给 map_async，以及更新对象的 numValue 属性和平方值的函数“calculate”。 然后“计算”函数返回具有更新值的对象。

一些代码：

import time
import multiprocessing
import random

class NumberHolder():
    def __init__(self,numValue):
        self.numValue = numValue    #Only one attribute

def calculate(obj):
    if random.random() >= 0.5:
        startTime = time.time()
        timeWaster = [random.random() for x in range(5000000)] #Waste time.
        endTime = time.time()           #Establish end time
        print("%d object got stuck in here for %f seconds"%(obj.numValue,endTime-startTime))

#Main Process
if __name__ == '__main__':
    numbersToSquare = [x for x in range(0,100)]     #I'm 
    taskList = []

    for eachNumber in numbersToSquare:
        taskList.append(NumberHolder(eachNumber))   #Create a list of objects whose numValue is equal to the numbers we want to square

    results = [] #Where the results will be stored
    pool = multiprocessing.Pool(processes=(multiprocessing.cpu_count() - 1)) #Don't use all my processing power.
    r = pool.map_async(calculate, taskList, callback=results.append)  #Using fxn "calculate", feed taskList, and values stored in "results" list
    r.wait()                # Wait on the results from the map_async

results = results[0]    #All of the entries only exist in the first offset
for eachObject in results:      #Loop through them and show them
    print(eachObject.numValue)          #If they calc'd "out of order", I'd expect append out of order

我发现了这个写得很好的回复，它似乎支持 map_async 可以产生“无序”结果的想法：multiprocessing.Pool: When to use apply, apply_async or map?。我还在这里查看了文档（http://docs.python.org/3.3/library/multiprocessing.html）。对于 map_async 它说这个方法“......如果指定了回调，那么它应该是一个接受单个参数的可调用。当结果准备好时，回调将应用于它（除非调用失败）。回调应该立即完成，否则处理结果的线程会被阻塞”

我是否误解了这应该如何工作？非常感谢任何帮助。

Answer 1

这是预期的行为。文档说：

map() 方法的变体，它返回一个结果对象。

“结果对象”只是一个保存计算结果的容器类。当您调用r.wait() 时，您将等待所有结果汇总并整理好。即使它不按顺序处理任务，结果仍然会按照原来的顺序。

如果您希望在计算结果时产生结果，请使用imap_unordered。