如何监控蟒蛇的concurrent.futures.ProcessPoolExecutor？

Question

我们在异步接收请求并在流程池中执行实际的同步处理的服务中使用来自concurrent.futures的concurrent.futures。

当我们遇到进程池耗尽的情况时，新的请求必须等到其他进程完成。

是否有一种方法可以查询进程池的当前用法？这将使我们能够监测他们的状况，并进行适当的能力规划。

如果没有，是否有支持这种监视/容量规划的异步接口的好的替代流程池实现？

Answer 1

最简单的方法是用期望的行为扩展ProcessPoolExecutor。下面的示例维护stdlib接口，不访问实现详细信息：

from concurrent.futures import ProcessPoolExecutor


class MyProcessPoolExecutor(ProcessPoolExecutor):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._running_workers = 0

    def submit(self, *args, **kwargs):
        future = super().submit(*args, **kwargs)
        self._running_workers += 1
        future.add_done_callback(self._worker_is_done)
        return future

    def _worker_is_done(self, future):
        self._running_workers -= 1

    def get_pool_usage(self):
        return self._running_workers

Answer 2

我最近以一种稍微不同的方式为自己解决了这个问题。简化了，以下是我所做的：

• 我在主循环范围内定义的一组外部跟踪待定期货。
• 我在每个未来都附加一个回调，这个回调是对一组期货的结束，允许它在完成后将未来从集合中删除。

因此，假设done()是在其他地方定义的实际回调函数，那么在我的主循环的范围中定义了以下内容：

bag = set()

def make_callback(b):

    def callback(f):
        nonlocal b
        b.remove(f)
        done(f)

    return callback

对于将来提交给ProcessPoolExecutor的每一个ProcessPoolExecutor，我都会添加回调：

f.add_done_callback(make_callback(bag))

在任何时候，都可以通过查看bag的内容来查看待定和正在运行的期货列表，可以选择地根据未来的running()方法的结果进行筛选。例如：

print(*bag, sep='\n')
print('running:', *(f for f in bag if f.running()))

对于许多简单的用例，模块级的set变量可能与闭包一样工作。