python - `multiprocessing` `starmap_async` 只调用回调一次？

Question

我有以下代码，它为 4 个工作线程创建一个池，并调用一个工作线程方法。该代码在大多数情况下都可以正常工作。运行时，我看到不同的工作人员被调用来处理工作。然而，当所有工作人员完成时，calc_completed 永远不会在最后调用一次。这是预期的行为吗？我预计回调会在每个工作人员完成时发生。

def calculate_worker(x, y):
    print 'working...'
    ...

def calc_completed(result):
    print 'completed: %s'%str(result)

def calc_errored(result):
    print 'error: %s'%str(result)

if __name__ == '__main__':  
    start, stop, step = 1, 1000, 1
    ranges = [(n, min(n+step, stop)) for n in xrange(start, stop, step)]

    pool = mp.Pool(processes=8)

    res = pool.starmap_async(calculate_worker, ranges,
                             callback=calculate_worker, error_callback=calc_completed)  

    pool.close()
    pool.join()
    d = res.get()       
    print(d)

Answer 1

calc_completed 仅在执行映射函数(此处:calculate_worker)时遇到任何错误时才会被调用。

代码中的另一个问题是您并行运行 calculate_worker 函数并将其用作回调。这没有多大意义，因为 calculate_worker 将被调用两次 - 第一次:作为 worker 函数，第二次:作为报告计算已完成的函数。你应该有两个不同的功能。

鉴于您提供的代码片段中的功能，我将按以下方式更改它:

res = pool.starmap_async(calculate_worker, ranges,
                         callback=calc_completed, 
                         error_callback=calc_errored)  

如果您想测试 calc_errored 是否被正确调用，那么您可以在 calculate_worker 函数中引入一些随机错误来查看它是否会被处理，例如

def calculate_worker(x, y):
    if (x % 7):
      x / (y - y)  # division by zero
    print 'working...'