正确设计 Multiprocessing.Manager 自定义对象

Question

我想使用 multiprocessing.Manager() 对象，这样我就可以将信息从工作人员异步发送到管理器以将信息发送到服务器。我拥有大约 10 个将 PDF 写入磁盘的实例。然后我想使用多处理包中的管理器对象将该数据发送到我的 S3 存储桶，因为我不想阻止本地内容生成。

所以我想知道如果我创建一个自定义管理器对象，这是正确的方法吗？提交给管理器对象的每个进程都会排队吗？或者如果我调用多个上传，经理会放弃一些调用吗？

下面是我想做的示例代码：

from multiprocessing.managers import BaseManager

class UploadClass(object):
    def upload(self, filePath, params, destUrl):
        # do stuff
        return results

class MyManager(BaseManager):
    pass

MyManager.register('uploads', UploadClass)

if __name__ == '__main__':
    manager = MyManager()
    manager.start()
    upload = manager.uploads()
    # do this wait for completion or do they perform this async
    print upload.upload(r"< path >", {...}, "some url")
    print upload.upload(r"< path >", {...}, "some url")

Answer 1

直接回答您的一些问题：

提交给管理器对象的每个进程都会排队吗？

Manager 服务器会生成一个新线程来处理每个传入的请求，因此您的所有请求都会立即开始处理。你可以在multiprocessing/managers.py里面看到这个：

def serve_forever(self):
    '''
    Run the server forever
    '''
    current_process()._manager_server = self
    try:
        try:
            while 1:
                try:
                    c = self.listener.accept()
                except (OSError, IOError):
                    continue
                t = threading.Thread(target=self.handle_request, args=(c,))
                t.daemon = True
                t.start()
        except (KeyboardInterrupt, SystemExit):
            pass
    finally:
        self.stop = 999
        self.listener.close()

如果我调用多个上传，经理会放弃一些调用吗？

不，不会掉线。

# do this wait for completion or do they perform this async
print upload.upload(r"< path >", {...}, "some url")
print upload.upload(r"< path >", {...}, "some url")

对upload.upload 的两个调用都是同步的；在UploadClass.upload 完成之前，他们不会返回。但是，如果您有多个脚本/线程/进程同时调用 upload.upload，则每个唯一调用都将同时发生在 Manager 服务器进程中其自己的线程内。

还有你最重要的问题：

这是正确的做法吗？

如果我正确理解了这个问题，我会说不。如果您只有一个脚本，然后在该脚本中生成十个 multiprocessing.Process 实例来写出 PDF，那么您应该只使用另一个 multiprocessing.Process 来处理上传：

def upload(self, q):
    for payload in iter(q.get, None):  # Keep getting from the queue until a None is found
        filePath, params, destUrl = payload
        # do stuff

def write_pdf(pdf_file_info, q):
   # write a pdf to disk here
   q.put((filepath, params, destUrl))  # Send work to the uploader
   # Move on with whatever comes next.

if __name__ == '__main__':
    pdf_queue = multiprocessing.Queue()

    # Start uploader
    upload_proc = multiprocessing.Process(upload, args=(pdf_queue,))
    upload_proc.start()

    # Start pdf writers
    procs = []
    for pdf in pdfs_to_write: 
         p = multiprocessing.Process(write_pdf, args=(pdf, pdf_queue))
         p.start()
         p.append(procs)

    # Wait for pdf writers and uploader to finish.
    for p in procs:
        p.join()
    pdf_queue.put(None) # Sending None breaks the for loop inside upload
    upload_proc.join()

如果您实际上对并发上传没问题，那么根本不需要单独的upload 进程 - 只需直接从 pdf 编写进程上传即可。

不过，从您的问题中很难判断这是否正是您正在做的事情。澄清后，我将调整最后一部分以适合您的特定用例。