优雅地处理子进程关闭

Question

我正在从事一个拥有大量工人的项目。我没有使用内置的multiprocessing.Pool，而是创建了自己的进程池。

它的工作方式是我创建了两个 multiprocessing.Queue 实例 - 一个用于向工作人员发送工作任务，另一个用于接收返回的结果。

每个工人只是像这样坐在一个永久运行的循环中：

while True:
    try:
        request = self.request_queue.get(True, 5)
    except Queue.Empty:
        continue
    else:
        result = request.callable(*request.args, **request.kwargs)
        self.results_queue.put((request, result))

还有一些错误处理代码，但我把它留了下来以备不时之需。每个工作进程都将daemon 设置为1。

我希望正确关闭主进程和所有子工作进程。到目前为止我的经验（使用 Ctrl+C）：

• 在没有特殊实现的情况下，每个子进程都会因 KeyboardInterrupt 回溯而停止/崩溃，但主进程不存在并且必须被终止 (sudo kill -9)。
• 如果我为子进程实现信号处理程序，设置为忽略 SIGINT，主线程会显示 KeyboardInterrupt 跟踪信息，但无论哪种方式都不会发生任何事情。
• 如果我为子进程和主进程实现信号处理程序，可以看到在主进程中调用了信号处理程序，但调用sys.exit()似乎没有任何效果。

我正在寻找处理此问题的“最佳实践”方式。我还在某处读到关闭与Queues 和Pipes 交互的进程可能会导致它们与其他进程死锁（由于信号量和内部使用的其他东西）。

我目前的方法如下： - 找到一种方法向将终止其主循环的每个进程（使用单独的命令队列或类似的）发送内部信号。 - 为发送关闭命令的主循环实现信号处理程序。子进程将有一个子处理程序，将它们设置为忽略信号。

这是正确的方法吗？

Answer 1

您需要注意的是处理在您想要关闭时队列中有消息的可能性，因此您需要一种方法让您的进程干净地排空其输入队列。假设您的主进程会识别到是时候关闭了，您可以这样做。

1. 向每个工作进程发送一个哨兵。这是一条特殊的消息（通常是None），看起来永远不会像普通消息。在哨兵之后，刷新并关闭每个工作进程的队列。

2. 在您的工作进程中使用类似于以下伪代码的代码：

   while True:  # Your main processing loop
       msg = inqueue.dequeue()  # A blocking wait
       if msg is None:
           break
       do_something()
   outqueue.flush()
   outqueue.close()
   

如果有可能多个进程在inqueue 上发送消息，您将需要更复杂的方法。此示例取自 Python 3.2 或更高版本中 logging.handlers.QueueListener 中 monitor 方法的源代码，显示了一种可能性。

            """
            Monitor the queue for records, and ask the handler
            to deal with them.

            This method runs on a separate, internal thread.
            The thread will terminate if it sees a sentinel object in the queue.
            """
            q = self.queue
            has_task_done = hasattr(q, 'task_done')
            # self._stop is a multiprocessing.Event object that has been set by the
            # main process as part of the shutdown processing, before sending
            # the sentinel           
            while not self._stop.isSet():
                try:
                    record = self.dequeue(True)
                    if record is self._sentinel:
                        break
                    self.handle(record)
                    if has_task_done:
                        q.task_done()
                except queue.Empty:
                    pass
            # There might still be records in the queue.
            while True:
                try:
                    record = self.dequeue(False)
                    if record is self._sentinel:
                        break
                    self.handle(record)
                    if has_task_done:
                        q.task_done()
                except queue.Empty:
                    break