python - multiprocessing.Pool.map_async 似乎……什么都没做？

Question

因此，我正在开发一个应用程序，每次启动时都必须根据哈希列表检查约 50 GB 的数据。显然这需要并行化，我不希望应用程序在“正在加载...”屏幕上挂起一分半钟。

我正在使用 multiprocessing.Pool 的 map_async 来处理这个问题；主线程调用 map_async(checkfiles, path_hash_pairs, callback) 并提供一个回调，告诉它在发现不匹配时发出警告。

问题是……没有任何反应。使用我的任务管理器查看 Python 进程显示它们生成然后立即终止而不做任何工作。他们从不打印任何东西，当然也从不完成并调用回调。

这个缩小的例子也表现出同样的问题:

def printme(x):
    time.sleep(1)
    print(x)
    return x**2

if __name__ == "__main__":
    l = list(range(0,512))

    def print_result(res):
        print(res)

    with multiprocessing.Pool() as p:
        p.map_async(printme, l, callback=print_result)
    p.join()
    time.sleep(10)

运行它，然后……没有任何反应。将 map_async 换成 map 完全符合预期。

我是犯了一个愚蠢的错误还是什么？

Answer 1

让我们看看会发生什么:

您正在使用上下文管理器自动“关闭”Pool，但是，重要的是，如果您检查 Pool.__exit__ 的源代码，您将查找:

def __exit__(self, exc_type, exc_val, exc_tb):
    self.terminate()

它只是调用terminate 而不是close。所以你仍然需要显式关闭 Pool 然后 join 它。

with multiprocessing.Pool() as p:
    p.map_async(printme, l, callback=print_result)
    p.close()
    p.join()

但在这种情况下，使用上下文管理器是没有意义的，只需要使用一个普通的形式:

p = multiprocessing.Pool()
p.map_async(printme, l, callback=print_result)
p.close()
p.join()

为什么它与 map 一起使用？因为 map 将阻止 util 所有工作都已完成。