21

我对multiprocessing模块很陌生。我只是尝试创建以下内容：我有一个进程的工作是从 RabbitMQ 获取消息并将其传递给内部队列（multiprocessing.Queue）。然后我想做的是：当新消息进来时产生一个进程。它可以工作，但是在工作完成后它会留下一个僵尸进程，它没有被它的父进程终止。这是我的代码：

主要流程：

 #!/usr/bin/env python

 import multiprocessing
 import logging
 import consumer
 import producer
 import worker
 import time
 import base

 conf = base.get_settings()
 logger = base.logger(identity='launcher')

 request_order_q = multiprocessing.Queue()
 result_order_q = multiprocessing.Queue()

 request_status_q = multiprocessing.Queue()
 result_status_q = multiprocessing.Queue()

 CONSUMER_KEYS = [{'queue':'product.order',
                   'routing_key':'product.order',
                   'internal_q':request_order_q}]
 #                 {'queue':'product.status',
 #                  'routing_key':'product.status',
 #                  'internal_q':request_status_q}]

 def main():
     # Launch consumers
     for key in CONSUMER_KEYS:
         cons = consumer.RabbitConsumer(rabbit_q=key['queue'],
                                        routing_key=key['routing_key'],
                                        internal_q=key['internal_q'])
         cons.start()

     # Check reques_order_q if not empty spaw a process and process message
     while True:
         time.sleep(0.5)
         if not request_order_q.empty():
             handler = worker.Worker(request_order_q.get())
             logger.info('Launching Worker')
             handler.start()

 if __name__ == "__main__":
     main()

这是我的工人：

 import multiprocessing
 import sys 
 import time
 import base

 conf = base.get_settings()
 logger = base.logger(identity='worker')

 class Worker(multiprocessing.Process):

     def __init__(self, msg):
         super(Worker, self).__init__()
         self.msg = msg 
         self.daemon = True

     def run(self):
         logger.info('%s' % self.msg)
         time.sleep(10)
         sys.exit(1)

因此，在处理完所有消息后，我可以看到带有ps aux命令的进程。但我真的希望他们一旦完成就被终止。谢谢。

4

3 回答 3

13

有几件事：

  1. 确保父母joins的孩子，以避免僵尸。请参阅Python 多处理终止进程

  2. is_alive()您可以使用成员函数检查孩子是否仍在运行。请参阅http://docs.python.org/2/library/multiprocessing.html#multiprocessing.Process

于 2013-10-11T16:23:50.583 回答
13

使用multiprocessing.active_children优于Process.join. 该函数active_children清除自上次调用以来创建的任何僵尸active_children。该方法join等待选定的进程。在此期间，其他进程可以终止并成为僵尸，但父进程不会注意到，直到加入等待的方法。要查看此操作：

import multiprocessing as mp
import time


def main():
    n = 3
    c = list()
    for i in range(n):
        d = dict(i=i)
        p = mp.Process(target=count, kwargs=d)
        p.start()
        c.append(p)
    for p in reversed(c):
        p.join()
        print('joined')


def count(i):
    print(f'{i} going to sleep')
    time.sleep(i * 10)
    print(f'{i} woke up')


if __name__ == '__main__':
    main()

以上将创建 3 个进程，每个进程终止 10 秒。由于代码是，最后一个进程首先加入，所以其他两个较早终止的进程将成为僵尸 20 秒。您可以通过以下方式查看它们：

ps aux | grep Z

如果进程按照它们将终止的顺序等待，则不会有僵尸。去掉对函数的调用reversed来查看这个案例。然而，在实际应用中，我们很少知道孩子会终止的顺序，因此使用该方法multiprocessing.Process.join会导致一些僵尸。

替代方案active_children不会留下任何僵尸。在上面的示例中，将循环替换为for p in reversed(c):：

while True:
    time.sleep(1)
    if not mp.active_children():
        break

看看会发生什么。

于 2016-02-12T21:15:27.243 回答
2

使用 active_children。 multiprocessing.active_children

于 2015-04-15T23:15:04.583 回答