Python:等待concurrent.futures.ThreadPoolExecutor的所有futures

Question

我已经给了concurrent.futures.ThreadPoolExecutor一堆任务，我想等到它们全部完成后再继续处理流。我怎样才能做到这一点，而不必保存所有的未来并对它们进行调用wait？（我想对遗嘱执行人提起诉讼。）

Answer 1

打电话给^{}：

shutdown(wait=True)

Signal the executor that it should free any resources that it is using when the currently pending futures are done executing. Calls to Executor.submit() and Executor.map() made after shutdown will raise RuntimeError.

If wait is True then this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed.

但是，如果您在一个列表中跟踪您的未来，那么您可以避免使用^{}函数关闭executor以备将来使用：

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

Wait for the Future instances (possibly created by different Executor instances) given by fs to complete. Returns a named 2-tuple of sets. The first set, named done, contains the futures that completed (finished or were cancelled) before the wait completed. The second set, named not_done, contains uncompleted futures.

请注意，如果您不提供timeout，它将一直等到所有未来完成。

您也可以使用^{}代替，但是您必须迭代它。

Answer 2

巴库留的回答是正确的。只是为了扩大一点。众所周知，上下文管理器有__enter__和__exit__方法。这里是如何定义class Executor（ThreadPoolExecutor的基类

class Executor(object):

    # other methods

    def shutdown(self, wait=True):
        """Clean-up the resources associated with the Executor.

        It is safe to call this method several times. Otherwise, no other
        methods can be called after this one.

        Args:
            wait: If True then shutdown will not return until all running
                futures have finished executing and the resources used by the
                executor have been reclaimed.
        """
        pass

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown(wait=True)
        return False

实际上是ThreadPoolExecutor定义了shutdown方法

class ThreadPoolExecutor(_base.Executor):
    def shutdown(self, wait=True):
        with self._shutdown_lock:
            self._shutdown = True
            self._work_queue.put(None)
        if wait:
            for t in self._threads:
                t.join()