嵌套concurrent.futures.ThreadPoolExecutor

Question

我有一个程序,我目前正在使用concurrent.futures.ThreadPoolExecutor同时运行多个任务.这些任务通常受I/O限制,涉及对本地数据库和远程REST API的访问.但是,这些任务本身可以分成子任务,这也可以从并发中受益.

我希望在任务中使用concurrent.futures.ThreadPoolExecutor是安全的.我编写了一个玩具示例,它似乎有效:

import concurrent.futures


def inner(i, j):
    return i, j, i**j


def outer(i):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(inner, i, j): j for j in range(5)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results


def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(outer, i): i for i in range(10)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.extend(future.result())
    print(results)


if __name__ == "__main__":
    main()

尽管这个玩具示例似乎有效,但我还是有信心这是故意的.我希望它是,因为否则使用执行程序执行任意代码是不安全的,以防它也使用concurrent.futures来利用并发.

Answer 1

从其他线程生成线程绝对没有问题。你的情况也不例外。

但迟早，生成线程的开销会相当高，并且生成更多线程实际上会导致您的软件变慢。

我强烈建议使用像asyncio这样的库，它可以完美地异步处理任务。它通过使用一个具有非阻塞 io 的线程来实现这一点。结果可能会比普通线程更快，因为开销要小得多。

如果您不想使用 asyncio，为什么不在 main 中创建另一个池执行器，并将其传递给函数outer()呢？这样，最多可以有 10 个 (2x5) 个线程，而不是 25 个 (5x5) 个线程，这更合理吗？

您不能传递调用的同一main()执行程序，因为这可能会导致死锁（每个执行程序都等待另一个执行程序完成才能进行调度）。outer()outer()outer()outer()inner()