谢谢Patrick
有了提示,我能够通过将Futures添加到列表中并手动调整队列大小来取消每个进程。如果没有这个提示,仍然会启动太多的进程。
似乎没有API可以调整队列大小、暂停执行或删除进程队列。
然而,实现这一点的唯一方法是在线程中运行主对象,以便主脚本可以随时终止它。我仍在尝试捕获“CancelledError”。
对我来说看起来很“不干净”,也不符合Pythonic的风格。我会接受任何其他建议。非常感谢。
from concurrent.futures import ProcessPoolExecutor, CancelledError
from time import sleep
from random import randint
from threading import Thread
def something_fancy():
sleep(randint(0, 5))
return 'im back!'
class Work:
def __init__(self):
self.exe = ProcessPoolExecutor(4)
self.futures = []
self.max_queue = 50
self.killed = False
def start_procs(self):
for i in range(200000):
while not self.killed:
if len(self.futures) <= self.max_queue:
t = self.exe.submit(something_fancy)
t.add_done_callback(self.done)
self.futures.append(t)
break
def done(self, f):
print f.result()
self.futures.remove(f)
def kill(self):
self.killed = True
for future in self.futures:
try:
future.cancel()
except CancelledError, e:
print e
if __name__ == '__main__':
work_obj = Work()
Thread(target=work_obj.start_procs).start()
sleep(5)
work_obj.kill()
编辑
from concurrent.futures import ProcessPoolExecutor, CancelledError
from time import sleep
from random import randint
from threading import Thread
def something_fancy():
sleep(0.5)
return 'Hello World, Future was running!'
class Work:
def __init__(self):
cpu_usage = 4
self.exe = ProcessPoolExecutor(cpu_usage)
self.futures = []
self.max_queue = cpu_usage*3
self.stop = False
self.paused = False
def start_procs(self):
for i in range(200000):
while not self.stop:
if len(self.futures) <= self.max_queue:
if not self.paused:
t = self.exe.submit(something_fancy)
t.add_done_callback(self._done)
self.futures.append(t)
break
def _done(self, f):
print f.result()
self.futures.remove(f)
def pause(self):
self.paused = False if self.paused else True
def shutdown(self):
self.stop = True
for future in self.futures:
try:
future.cancel()
except CancelledError, e:
print e
if __name__ == '__main__':
work_obj = Work()
Thread(target=work_obj.start_procs).start()
print 'Started'
sleep(5)
work_obj.pause()
print 'Paused'
sleep(5)
work_obj.pause()
print 'Continue'
sleep(5)
work_obj.shutdown()
print 'Shutdown'
这个可以用 - 但是仍然无法捕获CancelledError,而且还相当不干净。