使用 aiomultiprocess 在 Python 中进行多进程管理

Question

我对 Python 中的多处理有疑问。我需要创建异步进程，该进程运行时间未定义，进程数也未定义。新请求一到达，就必须使用请求中的规范创建一个新流程。我们使用 ZeroMQ 进行消息传递。还有一个 Process 是从一开始就开始，只有在整个脚本终止时才结束。

现在我正在寻找一种解决方案，如何等待所有进程，同时能够添加其他进程。

asyncio.gather()

这是我的第一个想法，但它在被调用之前需要进程列表。

class Object:
  def __init__(self, var):
     self.var = var

  async def run(self):
      *do async things*

class object_controller:
  
  def __init__(self):
     self.ctx = zmq.Context()
     self.socket = self.ctx.socket(zmq.PULL)
     self.socket.connect("tcp://127.0.0.1:5558")

     self.static_process = AStaticProcess()
     self.sp = aiomultiprocess.Process(target=self.static_process.run)
     self.sp.start()
     #here I need a good way to await this process


  def process(self, var):
    object = Object(var)
    process = aiomultiprocess.Process(target=object.run)
    process.start()
  
  def listener(self)
    while True:
      msg = self.socket.recv_pyobj()
      # here I need to find a way how I can start and await this process while beeing able to 
      # receive additional request, which result in additional processes which need to be awaited

这是一些希望能解释我的问题的代码。我需要一种等待进程的收集器。

初始化之后，对象和控制器之间没有交互，只有zeroMQ（静态进程和变量进程之间）。也没有回报。

Answer 1

如果您需要在同时等待新进程的同时启动进程，而不是显式调用await 来了解进程何时完成，请使用asyncio.create_task() 让它们在后台执行。这将返回一个Task 对象，该对象有一个add_done_callback 方法，您可以在该过程完成时使用它来做一些工作：

class Object:
  def __init__(self, var):
     self.var = var

  async def run(self):
      *do async things*

class object_controller:
  
  def __init__(self):
     self.ctx = zmq.Context()
     self.socket = self.ctx.socket(zmq.PULL)
     self.socket.connect("tcp://127.0.0.1:5558")

     self.static_process = AStaticProcess()
     self.sp = aiomultiprocess.Process(target=self.static_process.run)
     self.sp.start()
     asyncio.create_task(self.sp.join() self.handle_proc_finished)


  def process(self, var):
    object = Object(var)
    process = aiomultiprocess.Process(target=object.run)
    process.start()
  
  def listener(self)
    while True:
      msg = self.socket.recv_pyobj()
      process = aiomultiprocess.Process(...)
      process.start()
      t = asyncio.create_task(process.join())
      t.add_done_callback(self.handle_other_proc_finished)

  def handle_proc_finished(self, task):
     # do something

  def handle_other_proc_finished(self, task):
    # do something else

如果你想避免使用回调，你也可以传递 create_task 一个你自己定义的协程，它等待进程完成，然后做任何需要做的事情。

self.sp.start()
asyncio.create_task(wait_for_proc(self.sp))

async def wait_for_proc(proc):
   await proc.join()
   # do other stuff

Answer 2

您需要为流程创建任务列表或未来对象。此外，您不能在等待其他任务时将进程添加到事件循环中