在python中使用多进程multiprocessing

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

构造单个子进程

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

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.'

构造进程池

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

具体例子如下:

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。
具体例子如下:

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 '所有数据都写入并且读完'