multiprocessing.Process 产生的子进程如何正常退出？

reyleon

下面是一段示意代码 ttt.py：

1. #!/usr/bin/python
   
2. # -*- coding: utf-8 -*-
   
3. 
   
4. import multiprocessing
   
5. import time
   
6. import os
   
7. 
   
8. class test(multiprocessing.Process):
   
9.         def run(self):
   
10.                 print 'sub process [ppid]: %d,[pid]: %d -- sleep 5 --' %(os.getppid(),os.getpid())
    
11.                 time.sleep(5)
    
12.                 print 'sub process [ppid]: %d,[pid]: %d -- done --' % (os.getppid(),os.getpid())
    
13. 
    
14. b = 0
    
15. while True:
    
16.         print '-- [pid]: %d main process starting --' %os.getpid()
    
17.         while b < 10 :
    
18.                 p = test()
    
19.                 p.start()
    
20.                 b +=1
    
21.         print '-- [pid]: %d main process sleep 10 --' %os.getpid()
    
22.         time.sleep(100)

复制代码

运行后：

1. [root@localhost pytest]# python ttt.py
   
2. -- [pid]: 23695 main process starting --
   
3. sub process [ppid]: 23695,[pid]: 23696 -- sleep 5 --
   
4. sub process [ppid]: 23695,[pid]: 23697 -- sleep 5 --
   
5. sub process [ppid]: 23695,[pid]: 23698 -- sleep 5 --
   
6. sub process [ppid]: 23695,[pid]: 23699 -- sleep 5 --
   
7. sub process [ppid]: 23695,[pid]: 23700 -- sleep 5 --
   
8. sub process [ppid]: 23695,[pid]: 23702 -- sleep 5 --
   
9. sub process [ppid]: 23695,[pid]: 23701 -- sleep 5 --
   
10. sub process [ppid]: 23695,[pid]: 23703 -- sleep 5 --
    
11. sub process [ppid]: 23695,[pid]: 23704 -- sleep 5 --
    
12. -- [pid]: 23695 main process sleep 10 --
    
13. sub process [ppid]: 23695,[pid]: 23705 -- sleep 5 --
    
14. sub process [ppid]: 23695,[pid]: 23697 -- done --
    
15. sub process [ppid]: 23695,[pid]: 23696 -- done --
    
16. sub process [ppid]: 23695,[pid]: 23698 -- done --
    
17. sub process [ppid]: 23695,[pid]: 23699 -- done --
    
18. sub process [ppid]: 23695,[pid]: 23700 -- done --
    
19. sub process [ppid]: 23695,[pid]: 23701 -- done --
    
20. sub process [ppid]: 23695,[pid]: 23704 -- done --
    
21. sub process [ppid]: 23695,[pid]: 23703 -- done --
    
22. sub process [ppid]: 23695,[pid]: 23702 -- done --
    
23. sub process [ppid]: 23695,[pid]: 23705 -- done --

复制代码

产生了10个子进程，完成之后 ，我 ps -ef | grep python, 得到：

1. [root@localhost ~]# ps -ef | grep [p]ython
   
2. root     23695  8791  0 00:19 pts/4    00:00:00 python ttt.py
   
3. root     23696 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
4. root     23697 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
5. root     23698 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
6. root     23699 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
7. root     23700 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
8. root     23701 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
9. root     23702 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
   
10. root     23703 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
    
11. root     23704 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
    
12. root     23705 23695  0 00:19 pts/4    00:00:00 [python] <defunct>
    
13. [root@localhost ~]#
    

复制代码

子进程都是僵尸进程，怎么会这样呢？ 如何正确退出子进程？ python 新手，求解答！

bikong0411

#!/usr/bin/python
# -*- coding: utf-8 -*-

import multiprocessing
import time
import os

class test(multiprocessing.Process):
        def run(self):
                print 'sub process [ppid]: %d,[pid]: %d -- sleep 5 --' %(os.getppid(),os.getpid())
                time.sleep(5)
                print 'sub process [ppid]: %d,[pid]: %d -- done --' % (os.getppid(),os.getpid())

b = 0
while True:
        print '-- [pid]: %d main process starting --' %os.getpid()
        p={}
        for i in range(10) :
            p[i] = test()
            p[i].start()
            p[i].join()
        print '-- [pid]: %d main process sleep 10 --' %os.getpid()
        time.sleep(100)

timespace

start()
Start the process’s activity.

This must be called at most once per process object. It arranges for the object’s run() method to be invoked in a separate process.

join([timeout])
If the optional argument timeout is None (the default), the method blocks until the process whose join() method is called terminates. If timeout is a positive number, it blocks at most timeout seconds.

A process can be joined many times.

A process cannot join itself because this would cause a deadlock. It is an error to attempt to join a process before it has been started.

看文档，start下面就是join，回收子进程，和C系统调用wait/waitpid功能类似。

reyleon

回复 2# bikong0411


    这样的话是执行完一个，退出一个。 有没有办法，同时运行 10 个子进程呢 ？

reyleon

回复 3# timespace


    我的意图是：

while Ture:
   data = "从数据库里获取数据"
   for i in data:
        ＃然后
        p = class(i)
        p.start()
        p.join() 如果这里使用了这个，就停在这里了，得for循环里的子进程全部完成后while循环才能继续走。 我是希望产生子进程后，子进程在运行，但又不影响while循环继续。

但是这里不使用join。 子进程结束后就变成僵尸进程了，如我主题所述。

不知道咋搞了，能指点一二么？

reyleon

其实，我的需求是：

1.  不断从数据库获取数据，一次性可能是获取5条数据
2. 每条数据相当于一个任务，一个任务产生一个子进程去处理它，获取的任务立即执行，如立即执行5个任务
3. 主程序不等待任务（子进程）执行完，继续从数据库中获取数据（任务），如果获取到了，又立刻执行，如此反复

PS： 无需担心有多少条数据（任务）在运行！

---------------

但是，之前主程序衍生出的子进程，我现在不知道如何正确退出，不正确退出，就会变成僵尸进程了。

比如，以下的一个例子：

1. #!/usr/bin/python
   
2. # -*- coding: utf-8 -*-
   
3. 
   
4. import multiprocessing
   
5. import time
   
6. import os
   
7. 
   
8. class test(multiprocessing.Process):
   
9.         def run(self):
   
10.                 print 'sub process [ppid]: %d,[pid]: %d -- sleep 5 --' %(os.getppid(),os.getpid())
    
11.                 time.sleep(5)
    
12.                 print 'sub process [ppid]: %d,[pid]: %d -- done --' % (os.getppid(),os.getpid())
    
13. 
    
14. while True:
    
15.         i = 0
    
16.         p = {}
    
17.         print '-- [pid]: %d main process starting --' %os.getpid()
    
18.         while i < 10 :
    
19.                 p[i] = test()
    
20.                 p[i].start()
    
21.                 i+=1
    
22.         print '-- [pid]: %d main process sleep 100 --' %os.getpid()
    
23.         #sleep 100
    
24.         time.sleep(100)

复制代码

我执行后，是可以立即执行10个子进程的，而主进程会等待 100 秒 ，5秒钟后子进程都执行完了 ，主进程还在 sleep .
此时，我去 ps -ef 查询的时候，这些子进程就变成了僵尸进程拉。因为我没有正确退出子进程。

1. [root@localhost ~]# ps -ef | grep [p]ython
   
2. root      1406 14991  0 20:24 pts/1    00:00:00 python ttt.py
   
3. root      1407  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
4. root      1408  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
5. root      1409  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
6. root      1410  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
7. root      1411  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
8. root      1412  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
9. root      1413  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
   
10. root      1414  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
    
11. root      1415  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
    
12. root      1416  1406  0 20:24 pts/1    00:00:00 [python] <defunct>
    
13. [root@localhost ~]#

复制代码

两位大神，　我这个需求如何去做呢？　求指点！　@bikong0411　@timespace

timespace

回复 6# reyleon
这种需求用进程池或线程池更简单http://docs.python.org/2.7/library/multiprocessing.html#module-multiprocessing.pool，无需关心进程管理的问题。


   

reyleon

回复 7# timespace


主程序不等待任务（子进程）执行完，继续从数据库中获取数据（任务）！    能否帮我写个简单的示例？

timespace

回复 8# reyleon
如果你很确定，每条数据新建一个进程没有问题，比如进程数限制或服务器负载，那么也有一种办法。
基本思路：
1. 主进程管理子进程的创建和销毁
2. proxy进程负责数据IO操作
3. worker进程处理具体数据

下面是一个可正常运行的框架：

1. # -*- coding: utf-8 -*-
   
2. import multiprocessing as mp
   
3. import os
   
4. import random
   
5. from signal import signal, SIGINT, SIG_IGN, siginterrupt
   
6. import time
   
7. 
   
8. def data_source():
   
9.     """数据源。
   
10. 
    
11.     随机选择一个浮点数，作为worker进程的sleep时间，
    
12.     具体实践时可以将这部分实现改为读取数据库。
    
13.     """
    
14.     dataset = [0.1, 0.2, 0.3, 0.4, 0.5]
    
15.     while True:
    
16.         time.sleep(0.2)
    
17.         yield random.choice(dataset)
    
18. 
    
19. def proc_proxy(cntl_q, data_q, exit_flag):
    
20.     """从数据源读取数据。
    
21. 
    
22.     先通过cntl_q通知主进程，
    
23.     再将数据通过data_q发给worker。
    
24.     """
    
25.     for item in data_source():
    
26.         cntl_q.put({'event': 'data'})
    
27.         data_q.put(item)
    
28.         if exit_flag.is_set():
    
29.             cntl_q.put({'event': 'exit', 'pid': os.getpid()})
    
30.             break
    
31. 
    
32. 
    
33. def proc_worker(cntl_q, data_q):
    
34.     """处理数据。
    
35. 
    
36.     从data_q获取数据，处理完毕后通过cntl_q通知主进程，
    
37.     然后退出。
    
38.     """
    
39.     item = data_q.get()
    
40.     time.sleep(item)
    
41.     cntl_q.put({'event': 'exit', 'pid': os.getpid()})
    
42. 
    
43. def main():
    
44.     proc_pool = {} # 记录创建的所有子进程
    
45.     cntl_q = mp.Queue() # 控制信息传递队列
    
46.     data_q = mp.Queue() # 具体数据传递队列
    
47.     exit_flag = mp.Event() # 退出标记，初始值为False
    
48. 
    
49.     # 收到SIGINT，通知proxy停止读取数据
    
50.     signal(SIGINT, lambda x, y: exit_flag.set())
    
51.     siginterrupt(SIGINT, False)
    
52. 
    
53.     # 启动proxy进程，后续按需启动woker进程
    
54.     print 'main {} started'.format(os.getpid())
    
55.     proc = mp.Process(target=proc_proxy, args=(cntl_q, data_q, exit_flag))
    
56.     proc.start()
    
57.     proc_pool[proc.pid] = proc
    
58.     print 'proxy {} started'.format(proc.pid)
    
59. 
    
60.     while True:
    
61.         item = cntl_q.get()
    
62.         if item['event'] == 'data':
    
63.             proc = mp.Process(target=proc_worker, args=(cntl_q, data_q))
    
64.             proc.start()
    
65.             proc_pool[proc.pid] = proc
    
66.             print 'worker {} started'.format(proc.pid)
    
67.         elif item['event'] == 'exit':
    
68.             proc = proc_pool.pop(item['pid'])
    
69.             proc.join()
    
70.             print 'child {} stopped'.format(item['pid'])
    
71.         else:
    
72.             print 'It\'s impossible !'
    
73. 
    
74.         if not proc_pool: # 所有子进程均已退出
    
75.             break
    
76. 
    
77.     print 'main {} stopped'.format(os.getpid())
    
78. 
    
79. if __name__ == '__main__':
    
80.     main()

复制代码

reyleon

回复 9# timespace


    太感谢啦，经过实际测试，完美符合我的需求，而且加入了信号控制，很完善！！   灰常感谢！

    关于你说的进程控制，我是在数据库层面，查询任务的时候做好了控制的，最多不会超过查询的任务数，所以不用担心。