Python multiprocessing.Queue 未接收来自分叉进程的放置

Posted

技术标签:

【中文标题】Python multiprocessing.Queue 未接收来自分叉进程的放置【英文标题】:Python multiprocessing.Queue not receiving puts from forked processes 【发布时间】:2020-04-11 15:05:05 【问题描述】:

我正在创建固定数量的分叉子进程并尝试让它们通过multiprocessing.Queue 返回结果。这会导致一些意外行为。

import multiprocessing
import os

def main():
    n_workers = 4

    q = multiprocessing.Queue(n_workers)

    for i in range(n_workers):
        if os.fork() == 0:
            print(f"child i put i")
            q.put(i)
            print(f"child i exiting")
            os._exit(0)

    for i in range(n_workers):
        res = q.get()
        print(f"parent got res")

    print("parent exiting")


if __name__ == "__main__":
    main()

当我运行它时,所有子进程都将他们的结果排入队列并终止,但父进程挂起:

child 0 put 0                                                                              │
child 1 put 1                                                                              │
child 2 put 2                                                                              │
child 3 put 3                                                                              │
child 0 exiting                                                                            │
child 1 exiting                                                                            │
child 2 exiting                                                                            │
child 3 exiting                                                                            │
parent got 0

【问题讨论】:

【参考方案1】:

问题是在将数据放入队列后立即调用os._exit(0)

multiprocessing docs 解释如何将数据添加到队列中:

当一个对象被放入队列时,该对象被腌制,然后后台线程将腌制的数据刷新到底层管道。

因为进程是分叉的,所以必须调用os._exit(0)(而不是sys.exit(0)),但它是does not do any cleanup。如果后台线程还没有刷新数据,就会丢失!

解决方法是先调用close(),然后再调用join_thread()

import multiprocessing
import os

def main():
    n_workers = 4

    q = multiprocessing.Queue(n_workers)

    for i in range(n_workers):
        if os.fork() == 0:
            print(f"child i put i")
            q.put(i)
            print(f"child i exiting")

            q.close()  # indicate nothing else will be queued by this process
            q.join_thread()  # wait for the background thread to flush the data

            os._exit(0)

    for i in range(n_workers):
        res = q.get()
        print(f"parent got res")

    print("parent exiting")


if __name__ == "__main__":
    main()

【讨论】:

以上是关于Python multiprocessing.Queue 未接收来自分叉进程的放置的主要内容,如果未能解决你的问题,请参考以下文章

Python代写,Python作业代写,代写Python,代做Python

Python开发

Python,python,python

Python 介绍

Python学习之认识python

python初识