Python 池 - 当我遇到超时异常时如何保留 Python 进程 运行?
Python Pool - How do I keep Python Process running when I get a timeout exception?
我正在使用 Python 多处理库。每当其中一个进程抛出超时错误时,我的应用程序就会自行结束。我想保持流程正常运行。
我有一个订阅队列并监听传入消息的函数:
def process_msg(i):
#get new message from the queue
#process it
import time
time.sleep(10)
return True
我创建了一个 Pool,它创建了 6 个进程并执行上面的 process_msg() 函数。
当函数超时时,我希望 Pool 再次调用该函数并等待新消息而不是退出:
if __name__ == "main":
import multiprocessing
from multiprocessing import Pool
pool = Pool(processes=6)
collection = range(6)
try:
val = pool.map_async(process_msg, collection)
try:
res = val.get(5)
except TimeoutError:
print('timeout here')
pool.close()
pool.terminate()
pool.join()
代码运行,当我超时时,应用程序自行终止。
我想要它做的是打印发生的超时并再次调用相同的函数。
什么是正确的方法?
这是一个可以运行的程序的框架。您遇到的主要问题是 pool.terminate
的使用,它“在未完成未完成的工作的情况下立即停止工作进程”(请参阅 documentation)。
from multiprocessing import Pool, TimeoutError
def process_msg(i):
#get new message from the queue
#process it
import time
print(f"Starting to sleep, proxess # {i}")
time.sleep(10)
return True
def main():
print("in main")
pool = Pool(processes=6)
collection = range(6)
print("About to spawn sub processes")
val = pool.map_async(process_msg, collection)
while True:
try:
print("Waiting for results")
res = val.get(3)
print(f"Res is {res}")
break
except TimeoutError:
print("Timeout here")
print("Closing pool")
pool.close()
# pool.terminate() # do not terminate - it kill the child processes
print ("Joining pool")
pool.join()
print("exiting main")
if __name__ == "__main__":
main()
这段代码的输出是:
in main
About to spawn sub processes
Waiting for results
Starting to sleep, proxess # 0
Starting to sleep, proxess # 1
Starting to sleep, proxess # 2
Starting to sleep, proxess # 3
Starting to sleep, proxess # 4
Starting to sleep, proxess # 5
Timeout here
Waiting for results
Timeout here
Waiting for results
Timeout here
Waiting for results
Res is [True, True, True, True, True, True]
Closing pool
Joining pool
exiting main
我正在使用 Python 多处理库。每当其中一个进程抛出超时错误时,我的应用程序就会自行结束。我想保持流程正常运行。
我有一个订阅队列并监听传入消息的函数:
def process_msg(i):
#get new message from the queue
#process it
import time
time.sleep(10)
return True
我创建了一个 Pool,它创建了 6 个进程并执行上面的 process_msg() 函数。 当函数超时时,我希望 Pool 再次调用该函数并等待新消息而不是退出:
if __name__ == "main":
import multiprocessing
from multiprocessing import Pool
pool = Pool(processes=6)
collection = range(6)
try:
val = pool.map_async(process_msg, collection)
try:
res = val.get(5)
except TimeoutError:
print('timeout here')
pool.close()
pool.terminate()
pool.join()
代码运行,当我超时时,应用程序自行终止。
我想要它做的是打印发生的超时并再次调用相同的函数。
什么是正确的方法?
这是一个可以运行的程序的框架。您遇到的主要问题是 pool.terminate
的使用,它“在未完成未完成的工作的情况下立即停止工作进程”(请参阅 documentation)。
from multiprocessing import Pool, TimeoutError
def process_msg(i):
#get new message from the queue
#process it
import time
print(f"Starting to sleep, proxess # {i}")
time.sleep(10)
return True
def main():
print("in main")
pool = Pool(processes=6)
collection = range(6)
print("About to spawn sub processes")
val = pool.map_async(process_msg, collection)
while True:
try:
print("Waiting for results")
res = val.get(3)
print(f"Res is {res}")
break
except TimeoutError:
print("Timeout here")
print("Closing pool")
pool.close()
# pool.terminate() # do not terminate - it kill the child processes
print ("Joining pool")
pool.join()
print("exiting main")
if __name__ == "__main__":
main()
这段代码的输出是:
in main
About to spawn sub processes
Waiting for results
Starting to sleep, proxess # 0
Starting to sleep, proxess # 1
Starting to sleep, proxess # 2
Starting to sleep, proxess # 3
Starting to sleep, proxess # 4
Starting to sleep, proxess # 5
Timeout here
Waiting for results
Timeout here
Waiting for results
Timeout here
Waiting for results
Res is [True, True, True, True, True, True]
Closing pool
Joining pool
exiting main