我正在使用Python的multiprocessing模块进行科学并行处理。在我的代码中,我使用几个工作进程来执行重要任务,以及一个写入进程将结果保存到磁盘。要写入的数据通过队列从工作进程发送到写入进程。数据本身非常简单,仅包含一个包含文件名和两个浮点数列表的元组。经过数小时的处理后,写入进程经常会卡住。更具体地说,以下代码段:
while (True):
try:
item = queue.get(timeout=60)
break
except Exception as error:
logging.info("Writer: Timeout occurred {}".format(str(error)))
循环将永远不会退出,我会持续收到“超时”消息。
此外,我还实现了一个日志记录过程,其中输出了队列的状态等信息,尽管我会收到上面的超时错误消息,但对qsize()函数的调用始终返回完整的队列(在我的情况下为48)。
我已经彻底检查了队列对象的文档,并找不到可能解释为什么get()函数同时返回超时和队列已满的情况的可能性。
有任何想法吗?
编辑:
我修改了代码以确保我捕获了空队列异常:
while (True):
try:
item = queue.get(timeout=60)
break
except Empty as error:
logging.info("Writer: Timeout occurred {}".format(str(error)))
.get()
,如果队列仍然是满的,就绕过了这个问题,通常会起作用,但这并不是真正的解决方案。你找到解决方案了吗? - Jakamultiprocessing.Queue
而不是来自其他模块的Queue
。对吗? - azelcerget
而不是阻塞式的get
?这个写入进程可能是一个守护进程,当主进程结束时它也会结束,所以在get
上阻塞不会导致程序无法终止。或者你可以发送一个特殊的“哨兵”消息,比如None
,表示没有更多的消息了,如果你不想使用守护进程,它应该返回。