multiprocessing.Pipe([duplex])

返回2个连接对象(conn1, conn2),代表管道的两端,默认是双向通信.如果duplex=False,conn1只能用来接收消息,conn2只能用来发送消息.不同于os.open之处在于os.pipe()返回2个文件描述符(r, w),表示可读的和可写的

实例如下:

#!/usr/bin/python#coding=utf-8import osfrom multiprocessing import Process, Pipedef send(pipe):    pipe.send(['spam'] + [42, 'egg'])    pipe.close()def talk(pipe):    pipe.send(dict(name = 'Bob', spam = 42))    reply = pipe.recv()    print('talker got:', reply)if __name__ == '__main__':    (con1, con2) = Pipe()    sender = Process(target = send, name = 'send', args = (con1, ))    sender.start()    print "con2 got: %s" % con2.recv()#从send收到消息    con2.close()    (parentEnd, childEnd) = Pipe()    child = Process(target = talk, name = 'talk', args = (childEnd,))    child.start()    print('parent got:', parentEnd.recv())    parentEnd.send({x * 2 for x in 'spam'})    child.join()    print('parent exit')

 

输出如下：

con2 got: ['spam', 42, 'egg']('parent got:', {'name': 'Bob', 'spam': 42})('talker got:', set(['ss', 'aa', 'pp', 'mm']))parent exit

　　

multiprocessing中使用子进程概念

from multiprocessing import Process

可以通过Process来构造一个子进程

p = Process(target=fun,args=(args))

再通过p.start()来启动子进程

再通过p.join()方法来使得子进程运行结束后再执行父进程

from multiprocessing import Processimport 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.'

　　

 

在multiprocessing中使用pool

如果需要多个子进程时可以考虑使用进程池(pool)来管理

from multiprocessing import Pool

 

from multiprocessing import Poolimport os, timedef 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()    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创建子进程的方法与Process不同，是通过

p.apply_async(func,args=(args))实现，一个池子里能同时运行的任务是取决你电脑的cpu数量，如我的电脑现在是有4个cpu，那会子进程task0,task1,task2,task3可以同时启动，task4则在之前的一个某个进程结束后才开始。

代码中的p.close()是关掉进程池子，是不再向里面添加进程了，对Pool对象调用join()方法会等待所有子进程执行完毕，调用join()之前必须先调用close()，调用close()之后就不能继续添加新的Process了。

当时也可以是实例pool的时候给它定义一个进程的多少

如果上面的代码中p=Pool(5)那么所有的子进程就可以同时进行

多个子进程间的通信

多个子进程间的通信就要采用第一步中说到的Queue，比如有以下的需求，一个子进程向队列中写数据，另外一个进程从队列中取数据，

 

#coding:gbkfrom multiprocessing import Process, Queueimport os, time, random# 写数据进程执行的代码:def write(q):    for value in ['A', 'B', 'C']:        print 'Put %s to queue...' % value        q.put(value)        time.sleep(random.random())# 读数据进程执行的代码:def read(q):    while True:        if not q.empty():            value = q.get(True)            print 'Get %s from queue.' % value            time.sleep(random.random())        else:            breakif __name__=='__main__':    # 父进程创建Queue，并传给各个子进程：    q = Queue()    pw = Process(target=write, args=(q,))    pr = Process(target=read, args=(q,))    # 启动子进程pw，写入:    pw.start()        # 等待pw结束:    pw.join()    # 启动子进程pr，读取:    pr.start()    pr.join()    # pr进程里是死循环，无法等待其结束，只能强行终止:    print    print '所有数据都写入并且读完'

　　

 

关于上面代码的几个有趣的问题

if __name__=='__main__':        # 父进程创建Queue，并传给各个子进程：    q = Queue()    p = Pool()    pw = p.apply_async(write,args=(q,))        pr = p.apply_async(read,args=(q,))    p.close()    p.join()       print '所有数据都写入并且读完'

　　

如果main函数写成上面的样本，本来我想要的是将会得到一个队列，将其作为参数传入进程池子里的每个子进程，但是却得到

RuntimeError: Queue objects should only be shared between processes through inheritance

的错误，查了下，大意是队列对象不能在父进程与子进程间通信，这个如果想要使用进程池中使用队列则要使用multiprocess的Manager类

if __name__=='__main__':    manager = multiprocessing.Manager()    # 父进程创建Queue，并传给各个子进程：    q = manager.Queue()    p = Pool()    pw = p.apply_async(write,args=(q,))    time.sleep(0.5)    pr = p.apply_async(read,args=(q,))    p.close()    p.join()        print    print '所有数据都写入并且读完'

　　

这样这个队列对象就可以在父进程与子进程间通信，不用池则不需要Manager，以后再扩展multiprocess中的Manager类吧

关于锁的应用，在不同程序间如果有同时对同一个队列操作的时候，为了避免错误，可以在某个函数操作队列的时候给它加把锁，这样在同一个时间内则只能有一个子进程对队列进行操作，锁也要在manager对象中的锁

#coding:gbkfrom multiprocessing import Process,Queue,Poolimport multiprocessingimport 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:            breakif __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 '所有数据都写入并且读完'