多进程 Multiprocessing 和多线程 threading 类似, 他们都是在 python 中用来并行运算的. 不过既然有了 threading, 为什么 Python 还要出一个 multiprocessing 呢?

原因很简单, 就是用来弥补 threading 的一些劣势, 即threading的GIL机制无法让thread对于CPU密集型计算没有提速的效果。

1.  使用方法：

(1)和thread比较类似，添加一个进程的方法是：

import multiprocessing as mp

def job(a,d):

print('aaaaa')

if __name__ == '__main__':

#传入1,2

p1 = mp.Process(target=job,args=(1,2))

p1.start()

p1.join()

最终输出结果：

aaaaa

(2)存储进程输出Queue

Queue的功能是将每个核或线程的运算结果放在队里中， 等到每个线程或核运行完毕后再从队列中取出结果， 继续加载运算。原因很简单, 多线程或者多进程调用的函数不能有返回值, 所以使用Queue存储多个线程运算的结果。

import multiprocessing as mp

def job(q):

res = 0

for i in range(1000):

res+=i+i**2+i**3

q.put(res)

if __name__ == '__main__':

q = mp.Queue()

p1 = mp.Process(target=job,args=(q,))

p2 = mp.Process(target=job,args=(q,))

p1.start()

p2.start()

p1.join()

p2.join()

res1 = q.get()

res2 = q.get()

print(res1)

print(res2)

print(res1+res2)

(3)对比多线程、多进程、普通的运算时间

import multiprocessing as mp

import threading as td

import time

def job(q):

res = 0

for i in range(100000):

res = i+i**2+i**3

q.put(res) # queue

def multicore():

q = mp.Queue()

p1 = mp.Process(target=job,args=(q,))

p2 = mp.Process(target=job,args=(q,))

p1.start()

p2.start()

p1.join()

p2.join()

res1 = q.get()

res2 = q.get()

print('multicore:',res1+res2)

def multitd():

q = mp.Queue()

p1 = td.Thread(target=job,args=(q,))

p2 = td.Thread(target=job,args=(q,))

p1.start()

p2.start()

p1.join()

p2.join()

res1 = q.get()

res2 = q.get()

print('multitd:',res1+res2)

def normal():

res = 0

for _ in range(2):

for i in range(100000):

res = i+i**2+i**3

print('normal:',res)

if __name__ == '__name__':

st = time.time()

normal()

st1 = time.time()

print('The time consumption(normal):',st1-st)

multitd()

st2 = time.time()

print('The time consumption(multitd):',st2-st1)

multicore()

st3 = time.time()

print('The time consumption(multicore):',st3-st2)

显示结果

"""# range(1000000)

('normal:', 499999666667166666000000L)

('normal time:', 1.1306169033050537)

('thread:', 499999666667166666000000L)

('multithread time:', 1.3054230213165283)

('multicore:', 499999666667166666000000L)

('multicore time:', 0.646507978439331)"""

可看出，multicore的速度>普通>multitd的速度。

(4)进程池pool

Pool默认调用是CPU的核数，传入processes参数可自定义CPU核数

map() 放入迭代参数，返回多个结果

import multiprocessing as mp

def job(x):

return x*x

def multicore():

#定义一个pool

pool = mp.Pool()

res = pool.map(job,range(10))

print(res)

if __name__ == '__main__':

multicore()

返回结果：

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

[Finishedin 0.6s]

有了池子之后，就可以让池子对应某一个函数，我们向池子里丢数据，池子就会返回函数返回的值。 Pool和之前的Process的不同点是丢向Pool的函数有返回值，而Process的没有返回值。

接下来用map()获取结果，在map()中需要放入函数和需要迭代运算的值，然后它会自动分配给CPU核，返回结果。

3.自定义核数量

自定义核数量为3

def multicore():

pool = mp.Pool(processes=3) # 定义CPU核数量为3

res = pool.map(job, range(10))

print(res)

4.apply_async()只能放入一组参数，并返回一个结果，如果想得到map()的效果需要通过迭代

#应用apply_async()输出多个结果

multi_res = [pool.apply_async(job,(i,)) for i in range(10)]

#取出参数

print([res.get() for res in multi_res]

结果：

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

[0,1, 4, 9, 16, 25, 36, 49, 64, 81]

[Finishedin 0.9s]

(5)共享内存

这节我们学习如何定义共享内存。只有用共享内存才能让CPU之间有交流。

通过Value数据存储在一个共享的内存表中。

import multiprocessing as mp

value1 = mp.Value('i', 0)

value2 = mp.Value('d', 3.14)

其中：“i”是int型(带符号整型)；"d"是双精浮点型(double)

各参数代表的数据类型

(6)进程锁

没有进程锁

import multiprocessing as mp

import time

# 我们定义了一个共享变量v，两个进程都可以对它进行操作。

# 在job()中我们想让v每隔0.1秒输出一次累加num的结果，

# 但是在两个进程p1和p2 中设定了不同的累加值。所以接

# 下来让我们来看下这两个进程是否会出现冲突。

def job(v,num):

for _ in range(5):

time.sleep(0.1) #暂停0.1s

v.value += num #v.value获取共享变量值

print(v.value)

def multicore():

v = mp.Value('i',0) #定义共享变量

#一个迭代值是1，一个迭代值是3

p1 = mp.Process(target=job,args=(v,1))

p2 = mp.Process(target=job,args=(v,3))

#不加进程锁，看如何抢资源

p1.start()

p2.start()

p1.join()

p2.join()

if __name__ == '__main__':

multicore()

在上面的代码中，我们定义了一个共享变量v，两个进程都可以对它进行操作。 在job()中我们想让v每隔0.1秒输出一次累加num的结果，但是在两个进程p1和p2 中设定了不同的累加值。所以接下来让我们来看下这两个进程是否会出现冲突。

1

5

9

13

17

4

8

12

16

20

很明显，出现了进程冲突。

为了解决上述不同进程抢共享资源的问题，我们可以用加进程锁来解决。

import multiprocessing as mp

import time

# 我们定义了一个共享变量v，两个进程都可以对它进行操作。

# 在job()中我们想让v每隔0.1秒输出一次累加num的结果，

# 但是在两个进程p1和p2 中设定了不同的累加值。所以接

# 下来让我们来看下这两个进程是否会出现冲突。

#加入进程锁

def job(v,num,l):

l.acquire() # 锁住

for _ in range(5):

time.sleep(0.1) #暂停0.1s

v.value += num #v.value获取共享变量值

print(v.value)

l.release() #释放

def multicore():

v = mp.Value('i',0) #定义共享变量

l = mp.Lock()

#一个迭代值是1，一个迭代值是3

p1 = mp.Process(target=job,args=(v,1,l))

p2 = mp.Process(target=job,args=(v,3,l))

#不加进程锁，看如何抢资源

p1.start()

p2.start()

p1.join()

p2.join()

if __name__ == '__main__':

multicore()

结果：显然，进程锁保证了进程p1的完整运行，然后才进行了进程p2的运行

1

2

3

4

5

8

11

14

17

20[Finishedin 1.3s]