将multiprocessing.Process与concurrent.future._base.Future集成

Question

我需要创建子进程，使用Future接收结果，然后在需要时杀死其中一些。

为此，我已经子类化了multiprocessing.Process类，并从start（）方法返回一个Future对象。

问题是我无法在cb（）函数中接收结果，因为它永远不会被调用。

请帮助/建议是否可以通过其他方式完成此操作或我当前实施中缺少的内容？

以下是我目前的做法

from multiprocessing import Process, Queue
from concurrent.futures import _base
import threading
from time import sleep


def foo(x,q):
    print('result {}'.format(x*x))
    result = x*x
    sleep(5)
    q.put(result)


class MyProcess(Process):

    def __init__(self, target, args):
        super().__init__()
        self.target = target
        self.args = args
        self.f = _base.Future()

    def run(self):
        q = Queue()
        worker_thread = threading.Thread(target=self.target, args=(self.args+ (q,)))
        worker_thread.start()
        r = q.get(block=True)
        print('setting result {}'.format(r))
        self.f.set_result(result=r)
        print('done setting result')

    def start(self):
        f = _base.Future()
        run_thread = threading.Thread(target=self.run)
        run_thread.start()
        return f


def cb(future):
    print('received result in callback {}'.format(future))


def main():
    p1 = MyProcess(target=foo, args=(2,))
    f = p1.start()
    f.add_done_callback(fn=cb)

    sleep(10)


if __name__ == '__main__':

    main()

    print('Main thread dying')

Answer 1

在start方法中，您将创建一个新的Future，然后返回。这是一个与你设定结果的未来不同的未来，这个未来根本就没用过。尝试：

def start(self):
    run_thread = threading.Thread(target=self.run)
    run_thread.start()
    return self.f

但是，您的代码存在更多问题。您覆盖进程的start方法，将其替换为工作线程上的执行，因此实际上绕过了多处理。此外，您不应导入_base模块，这是从前导下划线看到的实现细节。你应该导入concurrent.futures.Future（它是同一个类，但通过公共API）。

这确实使用了多处理：

from multiprocessing import Process, Queue
from concurrent.futures import Future
import threading
from time import sleep


def foo(x,q):
    print('result {}'.format(x*x))
    result = x*x
    sleep(5)
    q.put(result)

class MyProcess(Process):

    def __init__(self, target, args):
        super().__init__()
        self.target = target
        self.args = args
        self.f = Future()

    def run(self):
        q = Queue()
        worker_thread = threading.Thread(target=self.target, args=(self.args+ (q,)))
        worker_thread.start()
        r = q.get(block=True)
        print('setting result {}'.format(r))
        self.f.set_result(result=r)
        print('done setting result')

def cb(future):
    print('received result in callback {}: {}'.format(future, future.result()))

def main():
    p1 = MyProcess(target=foo, args=(2,))
    p1.f.add_done_callback(fn=cb)
    p1.start()
    p1.join()
    sleep(10)

if __name__ == '__main__':
    main()
    print('Main thread dying')

而且你现在已经处于一个新的过程中，产生一个工作线程来执行你的目标函数应该不是必需的，你可以直接执行你的目标函数。如果目标函数引发了一个您不知道的异常，那么只有在成功时才会调用您的回调。所以如果你解决了这个问题，那么你就得到了：

from multiprocessing import Process
from concurrent.futures import Future
import threading
from time import sleep


def foo(x):
    print('result {}'.format(x*x))
    result = x*x
    sleep(5)
    return result

class MyProcess(Process):

    def __init__(self, target, args):
        super().__init__()
        self.target = target
        self.args = args
        self.f = Future()

    def run(self):
        try:
            r = self.target(*self.args)
            print('setting result {}'.format(r))
            self.f.set_result(result=r)
            print('done setting result')
        except Exception as ex:
            self.f.set_exception(ex)

def cb(future):
    print('received result in callback {}: {}'.format(future, future.result()))

def main():
    p1 = MyProcess(target=foo, args=(2,))
    p1.f.add_done_callback(fn=cb)
    p1.start()
    p1.join()
    sleep(10)

if __name__ == '__main__':
    main()
    print('Main thread dying')

这基本上是ProcessPoolExecutor所做的。