python之如何继承 multiprocessing.JoinableQueue

我正在尝试对 multiprocessing.JoinableQueue 进行子类化，这样我就可以跟踪被跳过而不是完成的作业。我正在使用 JoinableQueue 将作业传递给一组 multiprocessing.Process，并且我有一个 threading.Thread 填充队列。这是我的实现尝试:

import multiprocessing 
 
class InputJobQueue(multiprocessing.JoinableQueue): 
 
    def __init__(self, max_size): 
        super(InputJobQueue, self).__init__(0) 
        self._max_size = max_size 
        self._skipped_job_count = 0 
 
    def isFull(self): 
        return self.qsize() >= self._max_size 
 
    def taskSkipped(self): 
        self._skipped_job_count += 1 
        self.task_done() 

但是，我遇到了这个记录在案的问题 here :

class InputJobQueue(multiprocessing.JoinableQueue): 
TypeError 
: 
Error when calling the metaclass bases 
    function() argument 1 must be code, not str 

查看 multiprocessing 中的代码，我发现实际的类在 multiprocessing.queues 中。所以我尝试扩展那个类:

import multiprocessing.queues 
 
class InputJobQueue(multiprocessing.queues.JoinableQueue): 
 
    def __init__(self, max_size): 
        super(InputJobQueue, self).__init__(0) 
        self._max_size = max_size 
        self._skipped_job_count = 0 
 
    def isFull(self): 
        return self.qsize() >= self._max_size 
 
    def taskSkipped(self): 
        self._skipped_job_count += 1 
        self.task_done() 

但我得到的结果不一致:有时我的自定义属性存在，有时却不存在。例如。我的一个工作进程中报告了以下错误:

AttributeError: 'InputJobQueue' object has no attribute '_max_size' 

multiprocessing.JoinableQueue 的子类我缺少什么？

请您参考如下方法:

使用 multiprocessing，像 JoinableQueue 这样的对象在进程之间神奇地共享的方式是显式共享核心同步对象，并挑选“包装器”内容以传递一个管道。

如果您了解 pickling 的工作原理，可以查看来源 JoinableQueue并看到它正在使用 __getstate__/__setstate__。所以，你只需要覆盖那些来添加你自己的属性。像这样:

def __getstate__(self): 
    return super(InputJobQueue, self).__getstate__() + (self._max_size,) 
 
def __setstate__(self, state): 
    super(InputJobQueue, self).__setstate__(state[:-1]) 
    self._max_size = state[-1] 

我不保证这会真正工作，因为很明显这些类不是为子类设计的(针对您引用的错误的建议修复是记录这些类不能被子类化并找到一种使错误消息更好的方法……)。但它应该可以帮助您解决您在这里遇到的特定问题。