并发期货非阻塞地提交任务到处理池答案

【问题标题】：Concurrent futures submit task to process pool non-blockingly并发期货非阻塞地提交任务到处理池
【发布时间】：2017-02-11 02:35:19
【问题描述】：

这个问题是 Python 2 特有的，使用社区维护的 concurrent.futures 的反向移植。

我正在尝试使用 ProcessPoolExecutor（将 maxWorkers 设置为 2）并行运行两个任务。这些任务都是 Python 函数，我希望它们每个都在自己的进程中运行。他们不需要相互协调（我什至不需要知道退出状态）。我只想能够同时启动进程并限制在任何给定时刻并行运行的进程数量。

import concurrent.futures as futures
import time


def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)


pool = futures.ProcessPoolExecutor(max_workers=2)
pool.submit(do_stuff("a"))
print "a submitted!"
pool.submit(do_stuff("b"))

但是，这会打印

a 0 a 1 ... a 9 a submitted! b 0 b 1 ... b 9

为什么submit 是阻塞操作？是否有非阻塞等效项？

这是一个使用 multiprocessing 库的示例，它具有我想要的行为。它以非阻塞方式启动每个进程，然后调用join（可能只是waitpid(2) 的一个薄包装）。但是，这种技术无法让我限制在任何给定时刻并行运行的进程数量。

import multiprocessing
import time


def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)


proc_a = multiprocessing.Process(target=do_stuff, args="a")
proc_b = multiprocessing.Process(target=do_stuff, args="b")
proc_a.start()
proc_b.start()
proc_a.join()
proc_b.join()

【问题讨论】：

pool.submit(do_stuff("a")) 在这里，由于 do_stuff("a")，您实际上是在现场执行 do_stuff 方法

标签： python python-2.x concurrent.futures

【解决方案1】：

您的代码中的串行打印输出（而不是并发打印输出）似乎是由您用于提交函数参数的错误语法引起的。参数应使用逗号分隔函数Executor.submit()。试试这个版本吧。

import concurrent.futures as futures
import time

def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)

pool = futures.ProcessPoolExecutor(max_workers=2)
pool.submit(do_stuff, "a")
print "a submitted!"
pool.submit(do_stuff, "b")
print "b submitted!"

此外，我建议尽可能使用“with”语句来管理您的提交，因为这将确保正确关闭/关闭 concurrent.futures.Executor。文档中提到了这一点。

with futures.ProcessPoolExecutor(max_workers=2) as executor:
    executor.submit(do_stuff, "a") 
    print "a submitted!"
    executor.submit(do_stuff, "b")
    print "b submitted!"

【讨论】：