python concurrent.futures 跳过超时进程

Question

我正在处理成千上万的图像 url，并希望使用 concurrent.futures.ProcessPoolExecutor 来加快速度。

由于部分网址损坏或图片较大，处理功能可能会挂起或在处理过程中意外消耗大量时间。我想在处理函数上添加一个超时时间，比如 10 秒，以摆脱这些无效图像。

我尝试在futures .as_completed 中设置timeout 参数，TimeoutException 可以成功提升。但是，似乎主进程仍然会等到超时子进程完成。有没有办法立即杀死超时子进程并将下一个url放入池中？

from concurrent import futures

def process(url):
    ### Some time consuming operation
    return result


def main():
    urls = ['url1','url2','url3',...,'url100']
    with futures.ProcessPoolExecutor(max_workers=10) as executor:
        future_list = {executor.submit(process, url):url for url in urls}
        results = []
        try:
            for future in futures.as_completed(future_list, timeout=10):
                results.append(future.result())
        except futures._base.TimeoutException:
            print("timeout")
    print(results)
if __name__ == '__main__':
    main()

在上面的例子中，假设我有 100 个 url，其中 10 个是无效的，可能会花费很多时间，如何获取其余 90 个 url 的处理结果列表？

Answer 1

不适用于concurrent.futures 库。

开发了pebble 模块来克服这种限制。

from pebble import ProcessPool
from concurrent.futures import TimeoutError

with process.ProcessPool() as pool:
    future = pool.schedule(function, args=(1,2), timeout=5)

    try:
        result = future.result()  # blocks until results are ready
    except TimeoutError as error:
        print("Function took longer than %d seconds" % error.args[1])