`multiprocessing` `starmap_async` 只调用一次回调？答案

【问题标题】：`multiprocessing` `starmap_async` only calls callback once?`multiprocessing` `starmap_async` 只调用一次回调？
【发布时间】：2019-04-27 05:22:06
【问题描述】：

我有以下代码，它为 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)

【问题讨论】：

标签： python python-2.7 python-multiprocessing

【解决方案1】：

calc_completed 只有在映射函数的执行过程中遇到任何错误时才会被调用（这里：calculate_worker）。

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

鉴于您提供的 sn-p 中的功能，我将通过以下方式对其进行更改：

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...'

【讨论】：