在python中使用多进程multiprocessing

在日常开发中,我们经常遇到一些需要循环批处理的数据,如果处理完一个批次数据后再处理下一批次,这样对服务器是一种浪费,同时也让程序运行时间大大加长,那么如何在python中使用多进程呢?

构造单个子进程

当只需要少数确定子进程时,可以采用这种方式。
具体例子如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from multiprocessing import Process
import os
# 子进程要执行的代码
def run_proc(name):
print "Run child process %s (%s)..." % (name, os.getpid())
if __name__=='__main__':
print "Parent process %s." % os.getpid()
p = Process(target=run_proc, args=('test',)) #构造子进程,指定目标方法和参数
print 'Process will start.'
p.start() #启动子进程
p.join() #当子进程运行完后再执行父进程
print 'Process end.'

构造进程池

当我们需要维持大量的子进程时,可以采用进程池的方式。它的好处是能够管理子进程的数量,统一管理子进程等等。

具体例子如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from multiprocessing import Pool
import os, time
def long_time_task(name):
print "Run task %s (%s)..." % (name, os.getpid())
start = time.time()
time.sleep(3)
end = time.time()
print "Task %s runs %0.2f seconds." % (name, (end - start))
if __name__=='__main__':
print "Parent process %s." % os.getpid()
p = Pool(5)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print 'Waiting for all subprocesses done...'
p.close()
p.join()
print 'All subprocesses done.'

在上述代码中,我们构造了一个带参数的pool。这个参数决定了同时执行子进程的数量,如果不指定则按照cpu数量设定。close方法指这个pool不再接收新的进程执行请求。join方法和单个子进程的join是一个效果。

进程间通讯和加锁

在写多进程程序时,我们不可避免地要在进程间通信或加锁。python提供了一个可以在进程间通信的queue和lock。
具体例子如下:

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
from multiprocessing import Process,Queue,Pool
import multiprocessing
import os, time, random
# 写数据进程执行的代码:
def write(q,lock):
lock.acquire() #加上锁
for value in ['A', 'B', 'C']:
print 'Put %s to queue...' % value
q.put(value)
lock.release() #释放锁
# 读数据进程执行的代码:
def read(q):
while True:
if not q.empty():
value = q.get(False)
print 'Get %s from queue.' % value
time.sleep(random.random())
else:
break
if __name__=='__main__':
manager = multiprocessing.Manager()
# 父进程创建Queue,并传给各个子进程:
q = manager.Queue()
lock = manager.Lock() #初始化一把锁
p = Pool()
pw = p.apply_async(write,args=(q,lock))
pr = p.apply_async(read,args=(q,))
p.close()
p.join()
print
print '所有数据都写入并且读完'