Python多进程学习

发表于 | 分类于 ， ， | 浏览 次
字数统计: 1,175 字 | 阅读时长 ≈ 6 分钟

python学习中的多进程总结

多进程

跨平台实现多进程需要用到multiprocessing这个模块，这个模块中有两个类Process和Pool是常用的。

Process类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|  __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *,mon=None)
|      初始化类，target指向函数名，name参数可以为该进程设置名称，args是为target指向的函数传入的参数，且为tuple类型的
|
|  is_alive(self)
|      返回这个进程是否alive
|
|  join(self, timeout=None)
|      等待，直到子进程执行完毕
|
|  run(self)
|      Method to be run in sub-process; can be overridden in sub-class
|
|  start(self)
|      开启子进程
|
|  terminate(self)
|      结束进程
|
|  ----------------------------------------------------------------------
|  Data descriptors inherited from multiprocessing.process.BaseProcess:
|
|  __dict__
|      dictionary for instance variables (if defined)
|
|  __weakref__
|      list of weak references to the object (if defined)
|
|  authkey
|
|  daemon
|      返回该进程是否为守护进程
|
|  exitcode
|      返回进程的退出代码
|
|  ident
|      Return identifier (PID) of process or `None` if it has yet to start
|
|  name
|
|  pid
|      Return identifier (PID) of process or `None` if it has yet to start
``` 

## 用Process实现多进程：

![](https://blog-1252108140.cosbj.myqcloud.com/201803/1520576484.0351057.png)

# Pool类

如果想要创建多个进程，可以用到进程池来创建。

问题：为什么不用for循环来创建呢？

答案：当然可以啦，但是要注意join()要放到for循环外面，否则的话会执行完process_1才执行process_2的。

![](https://blog-1252108140.cosbj.myqcloud.com/201803/1520576514.0676877.png)

示范代码：

![](https://blog-1252108140.cosbj.myqcloud.com/201803/1520576526.8598888.png)

注意点：

* <1>.如果要开启的进程数大于进程池最大进程数的话，就要等待进程池中有空位了，后面的进程再进去

* <2>.进程池Pool的对象方法apply_async直接就启动了进程，不需要再用start()来启动了。

* <3>.调用close()后，就不能再添加新的process了，且join()必须在close()后面。

# 进程间的通信

## subprocess的使用

在Python中，我们通过标准库中的subprocess包来fork一个子进程，并运行一个外部的程序。

**subprocess.call(command,shell=True)：**fork一个子进程，并运行一个外部程序(命令)在本进程中显示结果。

![](https://blog-1252108140.cosbj.myqcloud.com/201803/1520576588.688439.png)

如果命令执行不成功，会返回一个异常，所以要用到try except来处理异常。

**output = subprocess.check_output(command)：**将外部命令执行结果放入变量，不在控制台输出

![](https://blog-1252108140.cosbj.myqcloud.com/201803/1520576624.2668383.png)

上图做了对比，对比subprocess.call和check_output的返回结果，call执行成功返回一个0，否则返回非零数，check_output返回命令执行的结果。

```python
su = subprocess.Popen(command,shell=True,stdout=subprocess.PIPE,stdin=subprocess.PIPE,stderr=subprocess.PIPE)

将输入，输出和err都用subprocess的管道来传输。调用：su.communicate([stdin])返回一个tuple或者用两个变量来接收(第一个变量接收stdout传来的值，第二个变量接收stderr传来的值)。

communicate()传参问题：如果运行的外部命令或者程序需要交互才用communicate来传参数，且Popen中stdin必须是subprocess的管道。也可以用su.stdin.write()来给外部程序传入参数，读取命令返回值也可以用su.stdout.read()

multiprocessing提供的Queue

实例化：q = Queue()

获取数据：q.get(True)

传入数据：q.put(value)

注意点

  • A.Queue()只能用于进程间的通信，子进程与父进程的通信试用subprocess比较好

  • B.Queue()不能用于进程池的进程间通信，进程池的进程间通讯要用multiprocessing.Manager().Queue()参考代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
#!/usr/bin/python
#coding=utf-8

from multiprocessing import Process,Pool,Queue,Manager
import os,time,random

def write(q,list,i):

	print("[+] This is Process_"+str(i)+" and pid is %s START"%os.getpid())
	for item in list:
		q.put(item)
		print("[+] Pid %s Put : %s"%(os.getpid(),item))
	print("[+] This is Process_"+i+" and pid is %s CLOSE"%os.getpid())


def test():
	print(os.getpid())

def read(q):
	print('[====] Process to Read , pid = %s'%os.getpid())
	while True:
		data = q.get(True)
		print("[=] data from Queue : %s"%str(data))

def main():
	manager = Manager()
	q = manager.Queue()
	p = Pool(5)
	listx = []
	listx.append([x for x in range(10)]) 
	listx.append([x for x in range(100,110)])
	listx.append([x for x in range(200,210)])
	listx.append([x for x in range(300,310)])
	listx.append([x for x in range(400,410)])
	for i in range(5):
		if i==4:
			pass
			p.apply_async(read,args=(q,))
		else:
			list = listx[i]
			#print(list)
			p.apply_async(write,args=(q,list,i))
	p.close()
	p.join()

if __name__ == '__main__':
	main()