深入了解Python并发编程

并发方式线程[Thread]多线程几乎是每一个程序猿在使用每一种语言时都会首先想到用于解决并发的工具JS程序员请回避使用多线程可以有效的利用CPU资源Python例外。然而多线程所带来的程序的复杂度也不可避免尤其是对竞争资源的同步问题。然而在python中由于使用了全局解释锁GIL的原因代码并不能同时在多核上并发的运行也就是说Python的多线程不能并发很多人会发现使用多线程来改进自己的Python代码后程序的运行效率却下降了这是多么蛋疼的一件事呀实际上使用多线程的编程模型是很困难的程序员很容易犯错这并不是程序员的错误因为并行思维是反人类的我们大多数人的思维是串行精神分裂不讨论而且冯诺依曼设计的计算机架构也是以顺序执行为基础的。所以如果你总是不能把你的多线程程序搞定恭喜你你是个思维正常的程序猿Python提供两组线程的接口一组是thread模块提供基础的低等级Low Level接口使用Function作为线程的运行体。还有一组是threading模块提供更容易使用的基于对象的接口类似于Java可以继承Thread对象来实现线程还提供了其它一些线程相关的对象例如TimerLock使用thread模块的例子12345importthreaddefworker():thread worker functionprintWorkerthread.start_new_thread(worker)使用threading模块的例子123456importthreadingdefworker():thread worker functionprintWorkertthreading.Thread(targetworker)t.start()或者Java Style12345678910importthreadingclassworker(threading.Thread):def__init__(self):passdefrun():thread worker functionprintWorkertworker()t.start()进程 Process由于前文提到的全局解释锁的问题Python下比较好的并行方式是使用多进程这样可以非常有效的使用CPU资源并实现真正意义上的并发。当然进程的开销比线程要大也就是说如果你要创建数量惊人的并发进程的话需要考虑一下你的机器是不是有一颗强大的心。Python的mutliprocess模块和threading具有类似的接口。12345678frommultiprocessingimportProcessdefworker():thread worker functionprintWorkerpProcess(targetworker)p.start()p.join()由于线程共享相同的地址空间和内存所以线程之间的通信是非常容易的然而进程之间的通信就要复杂一些了。常见的进程间通信有管道消息队列Socket接口TCP/IP等等。Python的mutliprocess模块提供了封装好的管道和队列可以方便的在进程间传递消息。Python进程间的同步使用锁这一点喝线程是一样的。另外Python还提供了进程池Pool对象可以方便的管理和控制线程。远程分布式主机 Distributed Node随着大数据时代的到临摩尔定理在单机上似乎已经失去了效果数据的计算和处理需要分布式的计算机网络来运行程序并行的运行在多个主机节点上已经是现在的软件架构所必需考虑的问题。远程主机间的进程间通信有几种常见的方式TCPIPTCPIP是所有远程通信的基础然而API比较低级别使用起来比较繁琐所以一般不会考虑远程方法调用 Remote Function Call[RPC]远程对象 Remote Object远程对象是更高级别的封装程序可以想操作本地对象一样去操作一个远程对象在本地的代理。远程对象最广为使用的规范CORBACORBA最大的好处是可以在不同语言和平台中进行通信。当让不用的语言和平台还有一些各自的远程对象实现例如Java的RMIMS的DCOMPython的开源实现有许多对远程对象的支持Dopy]Fnorb CORBAICEomniORB CORBAPyroYAMI消息队列 Message Queue比起RPC或者远程对象消息是一种更为灵活的通信手段常见的支持Python接口的消息机制有RabbitMQZeroMQKafkaAWS SQS BOTO在远程主机上执行并发和本地的多进程并没有非常大的差异都需要解决进程间通信的问题。当然对远程进程的管理和协调比起本地要复杂。Python下有许多开源的框架来支持分布式的并发提供有效的管理手段包括CeleryCelery是一个非常成熟的Python分布式框架可以在分布式的系统中异步的执行任务并提供有效的管理和调度功能。SCOOPSCOOP Scalable COncurrent Operations in Python提供简单易用的分布式调用接口使用Future接口来进行并发。Dispy相比起Celery和SCOOPDispy提供更为轻量级的分布式并行服务PPPP Parallel Python是另外一个轻量级的Python并行服务AsyncoroAsyncoro是另一个利用Generator实现分布式并发的Python框架当然还有许多其它的系统我没有一一列出另外许多的分布式系统多提供了对Python接口的支持例如Spark伪线程 PseudoThread还有一种并发手段并不常见我们可以称之为伪线程就是看上去像是线程使用的接口类似线程接口但是实际使用非线程的方式对应的线程开销也不存的。greenletgreenlet提供轻量级的coroutines来支持进程内的并发。greenlet是Stackless的一个副产品使用tasklet来支持一中被称之为微线程mircothread的技术这里是一个使用greenlet的伪线程的例子123456789101112131415fromgreenletimportgreenletdeftest1():print12gr2.switch()print34deftest2():print56gr1.switch()print78gr1greenlet(test1)gr2greenlet(test2)gr1.switch()运行以上程序得到如下结果125634伪线程gr1 switch会打印12然后调用gr2 switch得到56然后switch回到gr1打印34然后伪线程gr1结束程序退出所以78永远不会被打印。通过这个例子我们可以看出使用伪线程我们可以有效的控制程序的执行流程但是伪线程并不存在真正意义上的并发。eventletgevent和concurence都是基于greenlet提供并发的。eventleteventlet是一个提供网络调用并发的Python库使用者可以以非阻塞的方式来调用阻塞的IO操作。123456789101112importeventletfromeventlet.greenimporturllib2urls[http://www.google.com,http://www.example.com,http://www.python.org]deffetch(url):returnurllib2.urlopen(url).read()pooleventlet.GreenPool()forbodyinpool.imap(fetch, urls):print(got body,len(body))执行结果如下(got body, 17629)(got body, 1270)(got body, 46949)eventlet为了支持generator的操作对urllib2做了修改接口和urllib2是一致的。这里的GreenPool和Python的Pool接口一致。geventgevent和eventlet类似1234567importgeventfromgeventimportsocketurls[www.google.com,www.example.com,www.python.org]jobs[gevent.spawn(socket.gethostbyname, url)forurlinurls]gevent.joinall(jobs, timeout2)print[job.valueforjobinjobs]执行结果如下[206.169.145.226, 93.184.216.34, 23.235.39.223]concurenceconcurence是另外一个利用greenlet提供网络并发的开源库我没有用过大家可以自己尝试一下。实战运用通常需要用到并发的场合有两种一种是计算密集型也就是说你的程序需要大量的CPU资源;另一种是IO密集型程序可能有大量的读写操作包括读写文件收发网络请求等等。计算密集型对应计算密集型的应用我们选用著名的蒙特卡洛算法来计算PI值。基本原理如下蒙特卡洛算法利用统计学原理来模拟计算圆周率在一个正方形中一个随机的点落在1/4圆的区域红色点的概率与其面积成正比。也就该概率 p Pi RR 4 R R 其中R是正方形的边长圆的半径。也就是说该概率是圆周率的1/4, 利用这个结论只要我们模拟出点落在四分之一圆上的概率就可以知道圆周率了为了得到这个概率我们可以通过大量的实验也就是生成大量的点看看这个点在哪个区域然后统计出结果。基本算法如下12345frommathimporthypotfromrandomimportrandomdeftest(tries):returnsum(hypot(random(), random()) 1for_inrange(tries))这里test方法做了ntries次试验返回落在四分之一圆中的点的个数。判断方法是检查该点到圆心的距离如果小于R则是在圆上。通过大量的并发我们可以快速的运行多次试验试验的次数越多结果越接近真实的圆周率。这里给出不同并发方法的程序代码非并发我们先在单线程但进程运行看看性能如何123456789101112131415161718frommathimporthypotfromrandomimportrandomimporteventletimporttimedeftest(tries):returnsum(hypot(random(), random()) 1for_inrange(tries))defcalcPi(nbFutures, tries):tstime.time()resultmap(test, [tries]*nbFutures)ret4.*sum(result)/float(nbFutures*tries)spantime.time()-tsprinttime spend , spanreturnretprintcalcPi(3000,4000)多线程 thread为了使用线程池我们用multiprocessing的dummy包它是对多线程的一个封装。注意这里代码虽然一个字的没有提到线程但它千真万确是多线程。通过测试我们开jing心ya的发现果然不出所料当线程池为1是它的运行结果和没有并发时一样当我们把线程池数字设置为5时耗时几乎是没有并发的2倍我的测试数据从5秒到9秒。所以对于计算密集型的任务还是放弃多线程吧。123456789101112131415161718192021frommultiprocessing.dummyimportPoolfrommathimporthypotfromrandomimportrandomimporttimedeftest(tries):returnsum(hypot(random(), random()) 1for_inrange(tries))defcalcPi(nbFutures, tries):tstime.time()pPool(1)resultp.map(test, [tries]*nbFutures)ret4.*sum(result)/float(nbFutures*tries)spantime.time()-tsprinttime spend , spanreturnretif__name____main__:pPool()print(pi {}.format(calcPi(3000,4000)))多进程 multiprocess理论上对于计算密集型的任务使用多进程并发比较合适在以下的例子中进程池的规模设置为5修改进程池的大小可以看到对结果的影响当进程池设置为1时和多线程的结果所需的时间类似因为这时候并不存在并发当设置为2时响应时间有了明显的改进是之前没有并发的一半然而继续扩大进程池对性能影响并不大甚至有所下降也许我的Apple Air的CPU只有两个核