python的多进程编程(1)

这段时间要写一个程序用到多线程,,花了点时间研究了下python的多线程,结果十分沮丧,熟悉python的人都知道,python并发的不易,我手头上就3本python参考书,有2本就很直接的打击python多线程.”head first python”里’要避免的问题’章节中引用了twitter微博中的一句话,”有3种bug:你的bug,我的bug……还有线程.”“python编程实战”里第四章 ‘python的高级并发技术’ 开头一句话更是让我欲哭无泪:”进入21世纪后,开发者对并发编程兴趣大增.有三个因素助长了这一趋势.其一,java语言令并发编程变得更加普及;其二,拥有多核CPU的计算机几乎随处可见;其三,目前大部分语言都支持并发编程.”想当初,我也是个java程序员,喜欢python简洁的语法和哲学才转战python的, 因为java,所以多线程?这个逻辑让我欲哭无泪啊!但是有问题了还是要解决,在网上看了N多博客,各种方案,无非就是进程+协程,或者用pypy,jython等没有GIL的解释器,但是都是只言片语,没有系统的讲解和深入的研究.参考书上也没有讲太多,没办法,只有看官方文档,多线程的thread坑太多,不建议使用,threading也没什么好讲的,照着书敲几个demo就行了,多进程部分各个博客都写的不一样,让人摸不着北,于是自己翻译官方文档
其实我很懒的,都是看别人的博客,不顶不回,自己也不写博客,但是Python多线程,进程,协程部分没有一个系统,完整的资料,只有自己动手了
进入正题之前,吐槽一下,csdn等博客,很多博客质量之差,让人无法忍受,写个代码,不交代运行环境,是win?linux?32位,64位?python2还是python3?特别是学django的时候,django版本差异搞的人痛不欲生,很多博客都不交代开发环境,最后说一句:凡是没有注明运行环境的代码或解决方案都是*!
我的开发环境 win8.1x64,python2.7.9 32位(因为有时用tkinter写桌面软件,为了打包成exe后能在32位win上运行,所以用的32位),翻译的是2.7.10的python官方文档的16.6 multiprocessing,本人英语水平有限,有些翻译可能有错误,或者读起来不是很通顺,建议读者对照着官方文档阅读.如果有错误和需要改进的地方,欢迎指正.

16.6 multiprocessing – Process-based “threading” interface

new in version 2.6

16.6.1 Introduction

multiprocessing是一个使用和threading模块相似的API的支持多进程的包.multiprocessing包支持本地并发和远程并发,使用多进程代替线程可以有效的避免GIL的限制.所以,multiprocessing模块能让开发者充分利用多核CPU.它能运行在Unix和Windows上.

multiprocessing模块也引进了一些threading模块中没有的API.一个典型的例子是Pool对象,提供一个方便的函数的并行执行多个输入值、分配过程的输入数据(数据并行).下面的例子演示了一个普遍用法:在模块中定义这样一个函数以便子进程能够成功的导入该模块.这个并行数据的基本示例使用了Pool.

from multiprocessing import Pool


def f(x):
return x * x

if __name__ == '__main__':
    p = Pool(5)
print p.map(f, [1, 2, 3])

结果print到标准输入

[1, 4, 9]

16.6.1.1 The Process class

在multiprocessing里,通过这样的方式创建多进程:创建一个Process对象,然后调用它的start()方法.Process使用threading.Thread的(译者注:相似的)API.下面是一个关于多进程编程的小例子

from multiprocessing import Process


def f(name):
print 'hello', name

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

为了显示相关的每一个进程的ID,下面是一个扩展示例

from multiprocessing import Process
import os


def info(title):
print title
print 'module name:', __name__
if hasattr(os, 'getppid'):  # only available on Unix
print 'parent process:', os.getppid()
print 'process id:', os.getpid()


def f(name):
    info('function f')
print 'hello', name

if __name__ == '__main__':
    info('main line')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

注:至于为什么在win操作系统中, if __name__ == ’ __main__’ 这段话是必须的,请参考 Programming guidelines

译者注:读者可以不写main方法,直接在命令行调用试试,会报错,2.7.8的文档里有这个错误示例,报错信息也有,不知道为什么在2.7.10的文档里把这段去掉了

16.6.12 Exchanging objects between processes(进程间通信)

multiprocessing支持2种方式的进程间通信

Queues

Queue是Queue.Queue的near clone(译者注:不知道怎么翻译,最近的克隆?子类?),示例:

from multiprocessing import Process, Queue


def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
print q.get() # prints "[42, None, 'hello']"
    p.join()

Queues进程安全,线程安全

Pipes

Pipe() 方法返回2个连接对象,这2个连接对象是由默认为双向的pipe连接的(翻译的真丑,原文是这样的:The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way)), 示例:

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
print parent_conn.recv() # prints "[42, None, 'hello']"
    p.join()

16.6.1.3 进程间的同步控制

multiprocessing拥有和threading一样的同步原语.比如说,可以用锁来保证同一时间只有一个进程能打印到标准输出:

from multiprocessing import Process,Lock

def f(alock,i):
    alock.acquire()
print 'hello world', i
    alock.release()

if __name__ == '__main__':
    lock = Lock()

for num in range(10):
        Process(target=f, args=(lock,num)).start()

如果不用锁,不同的进程的输出就很可能会全部混淆(原文:get all mixed up)

译者注:代码中用alock做标识符,我是写的,原文是用小写的L,不符合PEP8规范,而且容易造成误会,所以我用alock.

16.6.1.4 进程间共享状态(Sharing state between processes)

就像上文提到的,当使用并发编程时,通常情况下要极力避免使用共享状态.尤其是使用多进程编程时.
然而,如果你真的需要一些共享数据,multiprocessing提供了几个方法

Shared memory

使用Value或Array可以将数据存储在能够共享的内存映射.例如:

from multiprocessing import Process, Value, Array


def f(n, a):
    n.value = 3.1415926
for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

print num.value
print arr[:]

输出:

3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

当创建num和arr时用到的’d’和’i’参数,是array模块使用的类型代码:’d’表示双精度浮点数,’i’表示有符号整数.这些共享变量是进程安全和线程安全的.
为了更灵活地使用共享内存,可以使用multiprocessing.sharedctypes模块,它支持在分配的共享内存中创建任意的ctypes对象

Server process

Manager()会返回一个manager对象,它控制一个服务进程,能包含许多python对象并能允许其他进程使用代理操作这些对象
Manager()返回的manager支持一下类型:list,dict,Namespace,Lock,RLock,Semaphore,BoundSemahpore,Condition,Event,Queue,Value,Array.示例:

from multiprocessing import Process, Manager

def f(adict, alist):
    adict[1] = '1'
    adict['2'] = 2
    adict[0.25] = None
    alist.reverse()

if __name__ == '__main__':
    manager = Manager()

    adict = manager.dict()
    alist = manager.list(range(10))

    p = Process(target=f, args=(adict, alist))
    p.start()
    p.join()

print adict
print alist

输出:

{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

server porcess managers比共享内存对象更灵活好用,因为它支持所有类型的对象.并且,一个manager能够通过网络被不同电脑上的进程共享.但是,使用manager比使用共享内存(shared memory)要慢.

秒客网