无法在 Python 3.5 中子类化多处理队列

Question

我的最终目标是将 stdout 从几个子进程重定向到一些队列，并在某个地方打印出来（也许在一个小 GUI 中）。

第一步是将class Queue 子Queue 到一个行为很像stdout 的对象中。但这就是我被困的地方。 Subclass 多处理 Queue 在 Python v3.5 中似乎是不可能的。

# This is a Queue that behaves like stdout
# Unfortunately, doesn't work in Python 3.5   :-(
class StdoutQueue(Queue):
    def __init__(self,*args,**kwargs):
        Queue.__init__(self,*args,**kwargs, ctx='')

    def write(self,msg):
        self.put(msg)

    def flush(self):
        sys.__stdout__.flush()

我在下面的 post 中找到了这个片段（可能 Python 3.5 当时还不存在）：Python multiprocessing redirect stdout of a child process to a Tkinter Text

在 Python v3.5 中，当子class多处理 Queue class 时，您偶然发现了奇怪的错误消息。我发现了两个描述该问题的错误报告：

https://bugs.python.org/issue21367

https://bugs.python.org/issue19895

我有两个问题：

假设我想坚持使用 Python v3.5 - 转到以前的版本并不是一个真正的选择。我可以使用什么解决方法以某种方式子class 多处理队列？
如果我升级到 Python v3.6，错误是否仍然存在？

编辑：

当您尝试子class 此处找到的 Queue class 时存在一个已知问题：

from multiprocessing import Queue    # <- known issue: you cannot subclass
                                     #    this Queue class, because it is
                                     #    not a genuine python class.

但以下应该有效：

from multiprocessing.queues import Queue   # <- from this Queue class, you
                                           #    should be able to make a
                                           #    subclass. But Python 3.5
                                           #    refuses :-(

遗憾的是，即使这样在 Python v3.5 中也不起作用。您收到以下错误：

    C:\Users\..\myFolder > python myTest.py

        Traceback (most recent call last):
            File "myTest.py", line 49, in <module>
              q = StdoutQueue()
            File "myTest.py", line 22, in __init__
              super(StdoutQueue,self).__init__(*args,**kwargs)
        TypeError: __init__() missing 1 required keyword-only argument: 'ctx'

编辑：

感谢 Darth Kotik 解决问题！这是完整的代码，更新了他的解决方案。现在可以了。

import sys
import time
import multiprocessing as mp
import multiprocessing.queues as mpq
from threading import Thread
from tkinter import *

'''-------------------------------------------------------------------'''
'''                SUBCLASSING THE MULTIPROCESSING QUEUE              '''
'''                                                                   '''
'''         ..and make it behave as a general stdout io               '''
'''-------------------------------------------------------------------'''
# The StdoutQueue is a Queue that behaves like stdout.
# We will subclass the Queue class from the multiprocessing package
# and give it the typical stdout functions.
#
# (1) First issue
# Subclassing multiprocessing.Queue or multiprocessing.SimpleQueue
# will not work, because these classes are not genuine
# python classes.
# Therefore, you need to subclass multiprocessing.queues.Queue or
# multiprocessing.queues.SimpleQueue . This issue is known, and is not
# the reason for asking this question. But I mention it here, for
# completeness.
#
# (2) Second issue
# There is another problem that arises only in Python V5 (and beyond).
# When subclassing multiprocessing.queues.Queue, you have to provide
# a 'multiprocessing context'. Not doing that, leads to an obscure error
# message, which is in fact the main topic of this question. Darth Kotik
# solved it.
# His solution is visible in this code:
class StdoutQueue(mpq.Queue):

    def __init__(self,*args,**kwargs):
        ctx = mp.get_context()
        super(StdoutQueue, self).__init__(*args, **kwargs, ctx=ctx)

    def write(self,msg):
        self.put(msg)

    def flush(self):
        sys.__stdout__.flush()


'''-------------------------------------------------------------------'''
'''                           TEST SETUP                              '''
'''-------------------------------------------------------------------'''

# This function takes the text widget and a queue as inputs.
# It functions by waiting on new data entering the queue, when it
# finds new data it will insert it into the text widget.
def text_catcher(text_widget,queue):
    while True:
        text_widget.insert(END, queue.get())

def test_child(q):
    # This line only redirects stdout inside the current process
    sys.stdout = q
    # or sys.stdout = sys.__stdout__ if you want to print the child to the terminal
    print('child running')

def test_parent(q):
    # Again this only redirects inside the current (main) process
    # commenting this like out will cause only the child to write to the widget
    sys.stdout = q
    print('parent running')
    time.sleep(0.5)
    mp.Process(target=test_child,args=(q,)).start()

if __name__ == '__main__':
    gui_root = Tk()
    gui_txt = Text(gui_root)
    gui_txt.pack()
    q = StdoutQueue()
    gui_btn = Button(gui_root, text='Test', command=lambda:test_parent(q),)
    gui_btn.pack()

    # Instantiate and start the text monitor
    monitor = Thread(target=text_catcher,args=(gui_txt,q))
    monitor.daemon = True
    monitor.start()

    gui_root.mainloop()

Answer 1

>>> import multiprocessing
>>> type(multiprocessing.Queue)
<class 'method'>
AttributeError: module 'multiprocessing' has no attribute 'queues'
>>> import multiprocessing.queues
>>> type(multiprocessing.queues.Queue)
<class 'type'>

正如您所见，multiprocessing.Queue 只是 multiprocessing.queues.Queue class 的构造方法。如果你想做一个 child class 就做 class MyQueue(multiprocessing.queues.Queue)

你可以看到这个方法的来源here

编辑：好的。我现在知道你的问题了。正如您在上面的 link 中看到的，multiprocessing.Queue 将 ctx 参数传递给 Queue。所以我设法通过 __init__ 方法自己完成它来使其工作。我不完全理解 BaseContext object 应该在哪里获得 _name 属性，所以我手动传递了它。

def __init__(self,*args,**kwargs):
    from multiprocessing.context import BaseContext
    ctx = BaseContext()
    ctx._name = "Name"
    super(StdoutQueue,self).__init__(*args,**kwargs, ctx=ctx)

EDIT2：原来的文档有一些关于上下文的信息here。所以不用像我那样手动创建它，你可以

import multiprocessing
ctx = multiprocessing.get_context()

它将创建适当的上下文 _name 设置（在您的特定情况下设置为 'fork'），您可以将其传递到您的队列。

无法在 Python 3.5 中子类化多处理队列

Cannot subclass multiprocessing Queue in Python 3.5

python

multiprocessing

python-3.x

python-multiprocessing