multiprocessing.Queue 的管道损坏错误

2022-01-12 00:00:00 python python-2.7 multiprocessing parallel-processing

问题描述

在 python2.7 中，multiprocessing.Queue 在从函数内部初始化时会引发错误.我提供了一个重现问题的最小示例.

In python2.7, multiprocessing.Queue throws a broken error when initialized from inside a function. I am providing a minimal example that reproduces the problem.

#!/usr/bin/python
# -*- coding: utf-8 -*-

import multiprocessing

def main():
    q = multiprocessing.Queue()
    for i in range(10):
        q.put(i)

if __name__ == "__main__":
    main()

抛出下面的断管错误

Traceback (most recent call last):
File "/usr/lib64/python2.7/multiprocessing/queues.py", line 268, in _feed
send(obj)
IOError: [Errno 32] Broken pipe

Process finished with exit code 0

我无法解释原因.我们不能从函数内部填充 Queue 对象肯定会很奇怪.

I am unable to decipher why. It would certainly be strange that we cannot populate Queue objects from inside a function.

解决方案

这里发生的是，当你调用 main() 时，它会创建 Queue，放入 10对象并结束函数，垃圾收集其内部的所有变量和对象，包括 Queue.但是您收到此错误是因为您仍在尝试发送 Queue 中的最后一个号码.

What happens here is that when you call main(), it creates the Queue, put 10 objects in it and ends the function, garbage collecting all of its inside variables and objects, including the Queue. BUT you get this error because you are still trying to send the last number in the Queue.

来自文档文档:

"当一个进程第一次将一个项目放入队列时，一个 feeder 线程是开始将对象从缓冲区传输到管道中."

"When a process first puts an item on the queue a feeder thread is started which transfers objects from a buffer into the pipe."

由于 put() 是在另一个 Thread 中进行的，它不会阻塞脚本的执行，并允许在完成之前结束 main() 函数队列操作.

As the put() is made in another Thread, it is not blocking the execution of the script, and allows to ends the main() function before completing the Queue operations.

试试这个:

#!/usr/bin/python
# -*- coding: utf-8 -*-

import multiprocessing
import time
def main():
    q = multiprocessing.Queue()
    for i in range(10):
        print i
        q.put(i)
    time.sleep(0.1) # Just enough to let the Queue finish

if __name__ == "__main__":
    main()

应该有一种方法可以加入队列或阻止执行，直到将对象放入Queue，您应该查看文档.

There should be a way to join the Queue or block execution until the object is put in the Queue, you should take a look in the documentation.

相关文章