python - 如何将 multiprocessing.Pool 实例传递给 apply_async 回调函数？

Question

这是我的素因数分解程序，我在 pool.apply_async(findK, args=(N,begin,end)) 中添加了一个回调函数，消息提示 素数分解是over 当因式分解结束时，它工作正常。

import math
import multiprocessing 

def findK(N,begin,end):
    for k in range(begin,end):
        if N% k == 0:
            print(N,"=" ,k ,"*", N/k)
            return True
    return False


def prompt(result):
    if result:
        print("prime factorization is over")


def mainFun(N,process_num):
    pool = multiprocessing.Pool(process_num)
    for i in range(process_num):
        if i ==0 :
            begin =2
        else:
            begin = int(math.sqrt(N)/process_num*i)+1
        end = int(math.sqrt(N)/process_num*(i+1))
        pool.apply_async(findK, args=(N,begin,end) , callback = prompt)    
    pool.close()
    pool.join()    

if __name__ == "__main__":
    N = 684568031001583853
    process_num = 16
    mainFun(N,process_num)

现在我想更改apply_async中的回调函数，将提示更改为关闭函数以杀死所有其他进程。

def prompt(result):
    if result:
        pool.terminate()

池实例未在提示范围内定义或传递到提示中。
pool.terminate()不能在提示功能中工作。
如何将 multiprocessing.Pool 实例传递给 apply_async'callback 函数？
(我已经以类格式完成，只是添加一个类方法并调用 self.pool.terminate 可以杀死所有其他进程，如何以函数格式完成工作？)

如果不设置pool为全局变量，是否可以将pool传入回调函数？

Answer 1

不支持向回调函数传递额外的参数。然而，你有很多优雅的方法来解决这个问题。

您可以将池逻辑封装到一个对象中:

class Executor:
    def __init__(self, process_num):
        self.pool = multiprocessing.Pool(process_num)

    def prompt(self, result):
        if result:
            print("prime factorization is over")
            self.pool.terminate()

    def schedule(self, function, args):
        self.pool.apply_async(function, args=args, callback=self.prompt)

    def wait(self):
        self.pool.close()
        self.pool.join() 


def main(N,process_num):
    executor = Executor(process_num)
    for i in range(process_num):
        ...
        executor.schedule(findK, (N,begin,end))   
    executor.wait()

或者您可以使用 concurrent.futures.Executor实现返回一个 Future 对象。您只需在设置回调之前将池附加到 Future 对象。

def prompt(future):
    if future.result():
        print("prime factorization is over")
        future.pool_executor.shutdown(wait=False)

def main(N,process_num):
    executor = concurrent.futures.ProcessPoolExecutor(max_workers=process_num)
    for i in range(process_num):
        ...
        future = executor.submit(findK, N,begin,end)
        future.pool_executor = executor
        future.add_done_callback(prompt)