python百万级并发,python做接口绝对并发

让计算机程序并发的运行是一个经常被讨论的话题,今天我想讨论一下Python下的各种并发方式。

并发方式

线程(Thread)

多线程几乎是每一个程序猿在使用每一种语言时都会首先想到用于解决并发的工具(JS程序员请回避),使用多线程可以有效的利用CPU资源(Python例外)。然而多线程所带来的程序的复杂度也不可避免,尤其是对竞争资源的同步问题。

然而在python中由于使用了全局解释锁(GIL)的原因,代码并不能同时在多核上并发的运行,也就是说,Python的多线程不能并发,很多人会发现使用多线程来改进自己的Python代码后,程序的运行效率却下降了,这是多么蛋疼的一件事呀!如果想了解更多细节,推荐阅读这篇文章。实际上使用多线程的编程模型是很困难的,程序员很容易犯错,这并不是程序员的错误,因为并行思维是反人类的,我们大多数人的思维是串行(精神分裂不讨论),而且冯诺依曼设计的计算机架构也是以顺序执行为基础的。所以如果你总是不能把你的多线程程序搞定,恭喜你,你是个思维正常的程序猿:)

Python提供两组线程的接口,一组是thread模块,提供基础的,低**(Low Level)接口,使用Function作为线程的运行体。还有一组是threading模块,提供更容易使用的基于对象的接口(类似于Java),可以继承Thread对象来实现线程,还提供了其它一些线程相关的对象,例如Timer,Lock

使用thread模块的例子

import thread

def worker():

 """thread worker function"""

 print 'Worker'

thread.start_new_thread(worker)

使用threading模块的例子

import threading

def worker():

 """thread worker function"""

 print 'Worker'

t = threading.Thread(target=worker)

t.start()

或者Java Style

import threading

class worker(threading.Thread):

 def __init__(self):

 pass

 def run():

 """thread worker function"""

 print 'Worker'

 

t = worker()

t.start()

进程 (Process)

由于前文提到的全局解释锁的问题,Python下比较好的并行方式是使用多进程,这样可以非常有效的使用CPU资源,并实现真正意义上的并发。当然,进程的开销比线程要大,也就是说如果你要创建数量惊人的并发进程的话,需要考虑一下你的机器是不是有一颗强大的心。

Python的mutliprocess模块和threading具有类似的接口。

from multiprocessing import Process



def worker():

 """thread worker function"""

 print 'Worker'

p = Process(target=worker)

p.start()

p.join()

由于线程共享相同的地址空间和内存,所以线程之间的通信是非常容易的,然而进程之间的通信就要复杂一些了。常见的进程间通信有,管道,消息队列,Socket接口(TCP/IP)等等。

Python的mutliprocess模块提供了封装好的管道和队列,可以方便的在进程间传递消息。

Python进程间的同步使用锁,这一点喝线程是一样的。

另外,Python还提供了进程池Pool对象,可以方便的管理和控制线程。

远程分布式主机 (Distributed Node)

随着大数据时代的到临,摩尔定理在单机上似乎已经失去了效果,数据的计算和处理需要分布式的计算机网络来运行,程序并行的运行在多个主机节点上,已经是现在的软件架构所必需考虑的问题。

远程主机间的进程间通信有几种常见的方式

  • TCP/IPTCP/IP是所有远程通信的基础,然而API比较低级别,使用起来比较繁琐,所以一般不会考虑
  • 远程方法调用 Remote Function CallRPC是早期的远程进程间通信的手段。Python下有一个开源的实现RPyC
  • 远程对象 Remote Object远程对象是更高级别的封装,程序可以想操作本地对象一样去操作一个远程对象在本地的代理。远程对象最广为使用的规范CORBA,CORBA最大的好处是可以在不同语言和平台中进行通信。当让不用的语言和平台还有一些各自的远程对象实现,例如Java的RMI,MS的DCOMPython的开源实现,有许多对远程对象的支持
    • Dopy
    • Fnorb (CORBA)
    • ICE
    • omniORB (CORBA)
    • Pyro
    • YAMI
  • 消息队列 Message Queue比起RPC或者远程对象,消息是一种更为灵活的通信手段,常见的支持Python接口的消息机制有
    • RabbitMQ
    • ZeroMQ
    • Kafka
    • AWS SQS + BOTO

在远程主机上执行并发和本地的多进程并没有非常大的差异,都需要解决进程间通信的问题。当然对远程进程的管理和协调比起本地要复杂。

Python下有许多开源的框架来支持分布式的并发,提供有效的管理手段包括:

  • Celery Celery是一个非常成熟的Python分布式框架,可以在分布式的系统中,异步的执行任务,并提供有效的管理和调度功能。参考这里
  • SCOOPSCOOP (Scalable COncurrent Operations in Python)提供简单易用的分布式调用接口,使用Future接口来进行并发。
  • Dispy相比起Celery和SCOOP,Dispy提供更为轻量级的分布式并行服务
  • PP PP (Parallel Python)是另外一个轻量级的Python并行服务, 参考这里
  • AsyncoroAsyncoro是另一个利用Generator实现分布式并发的Python框架,

当然还有许多其它的系统,我没有一一列出

另外,许多的分布式系统多提供了对Python接口的支持,例如Spark

伪线程 (Pseudo-Thread)

还有一种并发手段并不常见,我们可以称之为伪线程,就是看上去像是线程,使用的接口类似线程接口,但是实际使用非线程的方式,对应的线程开销也不存的。

  • greenlet greenlet提供轻量级的coroutines来支持进程内的并发。greenlet是Stackless的一个副产品,使用tasklet来支持一中被称之为微线程(mirco-thread)的技术,这里是一个使用greenlet的伪线程的例子
from greenlet import greenlet



def test1():

 print 12

 gr2.switch()

 print 34

 

def test2():

 print 56

 gr1.switch()

 print 78

 

gr1 = greenlet(test1)

gr2 = greenlet(test2)

gr1.switch()

运行以上程序得到如下结果:

12

56

34

伪线程gr1 switch会打印12,然后调用gr2 switch得到56,然后switch回到gr1,打印34,然后伪线程gr1结束,程序退出,所以78永远不会被打印。通过这个例子我们可以看出,使用伪线程,我们可以有效的控制程序的执行流程,但是伪线程并不存在真正意义上的并发。

eventlet,gevent和concurence都是基于greenlet提供并发的。

  • eventlet

eventlet是一个提供网络调用并发的Python库,使用者可以以非阻塞的方式来调用阻塞的IO操作。

import eventlet

from eventlet.green import urllib2



urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']



def fetch(url):

 return urllib2.urlopen(url).read()



pool = eventlet.GreenPool()



for body in pool.imap(fetch, urls):

 print("got body", len(body))

执行结果如下

('got body', 17629)

('got body', 1270)

('got body', 46949)

eventlet为了支持generator的操作对urllib2做了修改,接口和urllib2是一致的。这里的GreenPool和Python的Pool接口一致。

  • gevent

gevent和eventlet类似,关于它们的差异大家可以参考这篇文章

import gevent

from gevent import socket

urls = ['www.google.com', 'www.example.com', 'www.python.org']

jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls]

gevent.joinall(jobs, timeout=2)



print [job.value for job in jobs]

执行结果如下:

['206.169.145.226', '93.184.216.34', '23.235.39.223']
  • concurence

concurence是另外一个利用greenlet提供网络并发的开源库,我没有用过,大家可以自己尝试一下。

实战运用

通常需要用到并发的场合有两种,一种是计算密集型,也就是说你的程序需要大量的CPU资源;另一种是IO密集型,程序可能有大量的读写操作,包括读写文件,收发网络请求等等。

计算密集型

对应计算密集型的应用,我们选用著名的蒙特卡洛算法来计算PI值。基本原理如下

 

蒙特卡洛算法利用统计学原理来模拟计算圆周率,在一个正方形中,一个随机的点落在1/4圆的区域(红色点)的概率与其面积成正比。也就该概率 p = Pi * R*R /4 : R* R , 其中R是正方形的边长,圆的半径。也就是说该概率是圆周率的1/4, 利用这个结论,只要我们模拟出点落在四分之一圆上的概率就可以知道圆周率了,为了得到这个概率,我们可以通过大量的实验,也就是生成大量的点,看看这个点在哪个区域,然后统计出结果。

基本算法如下:

from math import hypot

from random import random



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))

这里test方法做了n(tries)次试验,返回落在四分之一圆中的点的个数。判断方法是检查该点到圆心的距离,如果小于R则是在圆上。

通过大量的并发,我们可以快速的运行多次试验,试验的次数越多,结果越接近真实的圆周率。

这里给出不同并发方法的程序代码

  • 非并发我们先在单线程,但进程运行,看看性能如何
from math import hypot

from random import random

import eventlet

import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 result = map(test, [tries] * nbFutures)

 

 ret = 4. * sum(result) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



print calcPi(3000,4000)
  • 多线程 thread为了使用线程池,我们用multiprocessing的dummy包,它是对多线程的一个封装。注意这里代码虽然一个字的没有提到线程,但它千真万确是多线程。通过测试我们开(jing)心(ya)的发现,果然不出所料,当线程池为1是,它的运行结果和没有并发时一样,当我们把线程池数字设置为5时,耗时几乎是没有并发的2倍,我的测试数据从5秒到9秒。所以对于计算密集型的任务,还是放弃多线程吧。
from multiprocessing.dummy import Pool



from math import hypot

from random import random

import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 p = Pool(1)

 result = p.map(test, [tries] * nbFutures)

 ret = 4. * sum(result) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



if __name__ == '__main__':

 p = Pool()

 print("pi = {}".format(calcPi(3000, 4000)))
  • 多进程 multiprocess理论上对于计算密集型的任务,使用多进程并发比较合适,在以下的例子中,进程池的规模设置为5,修改进程池的大小可以看到对结果的影响,当进程池设置为1时,和多线程的结果所需的时间类似,因为这时候并不存在并发;当设置为2时,响应时间有了明显的改进,是之前没有并发的一半;然而继续扩大进程池对性能影响并不大,甚至有所下降,也许我的Apple Air的CPU只有两个核?当心,如果你设置一个非常大的进程池,你会遇到 Resource temporarily unavailable的错误,系统并不能支持创建太多的进程,毕竟资源是有限的。
from multiprocessing import Pool



from math import hypot

from random import random

import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 p = Pool(5)

 result = p.map(test, [tries] * nbFutures)

 ret = 4. * sum(result) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



if __name__ == '__main__':

 print("pi = {}".format(calcPi(3000, 4000)))
  • gevent (伪线程)不论是gevent还是eventlet,因为不存在实际的并发,响应时间和没有并发区别不大,这个和测试结果一致。
import gevent

from math import hypot

from random import random

import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 jobs = [gevent.spawn(test, t) for t in [tries] * nbFutures]

 gevent.joinall(jobs, timeout=2)

 ret = 4. * sum([job.value for job in jobs]) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



print calcPi(3000,4000)
  • eventlet (伪线程)
from math import hypot

from random import random

import eventlet

import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 pool = eventlet.GreenPool()

 result = pool.imap(test, [tries] * nbFutures)

 

 ret = 4. * sum(result) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



print calcPi(3000,4000)
  • SCOOP

SCOOP中的Future接口符合PEP-3148的定义,也就是在Python3中提供的Future接口。

在缺省的SCOOP配置环境下(单机,4个Worker),并发的性能有提高,但是不如两个进程池配置的多进程。

from math import hypot

from random import random

from scoop import futures



import time



def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))



def calcPi(nbFutures, tries):

 ts = time.time()

 expr = futures.map(test, [tries] * nbFutures)

 ret = 4. * sum(expr) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



if __name__ == "__main__":

 print("pi = {}".format(calcPi(3000, 4000)))
  • Celery

任务代码

from celery import Celery



from math import hypot

from random import random

 

app = Celery('tasks', backend='amqp', broker='amqp://guest@localhost//')

app.conf.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite'

 

@app.task

def test(tries):

 return sum(hypot(random(), random()) < 1 for _ in range(tries))

客户端代码

from celery import group

from tasks import test



import time



def calcPi(nbFutures, tries):

 ts = time.time()

 result = group(test.s(tries) for i in xrange(nbFutures))().get()

 

 ret = 4. * sum(result) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



print calcPi(3000, 4000)

使用Celery做并发的测试结果出乎意料(环境是单机,4frefork的并发,消息broker是rabbitMQ),是所有测试用例里最糟糕的,响应时间是没有并发的5~6倍。这也许是因为控制协调的开销太大。对于这样的计算任务,Celery也许不是一个好的选择。

  • asyncoroAsyncoro的测试结果和非并发保持一致。
import asyncoro



from math import hypot

from random import random

import time



def test(tries):

 yield sum(hypot(random(), random()) < 1 for _ in range(tries))





def calcPi(nbFutures, tries):

 ts = time.time()

 coros = [ asyncoro.Coro(test,t) for t in [tries] * nbFutures]

 ret = 4. * sum([job.value() for job in coros]) / float(nbFutures * tries)

 span = time.time() - ts

 print "time spend ", span

 return ret



print calcPi(3000,4000)

IO密集型

IO密集型的任务是另一种常见的用例,例如网络WEB服务器就是一个例子,每秒钟能处理多少个请求时WEB服务器的重要指标。

我们就以网页读取作为最简单的例子

from math import hypot

import time

import urllib2



urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']



def test(url):

 return urllib2.urlopen(url).read()



def testIO(nbFutures):

 ts = time.time()

 map(test, urls * nbFutures)



 span = time.time() - ts

 print "time spend ", span



testIO(10)

在不同并发库下的代码,由于比较类似,我就不一一列出。大家可以参考计算密集型中代码做参考。

通过测试我们可以发现,对于IO密集型的任务,使用多线程,或者是多进程都可以有效的提高程序的效率,而使用伪线程性能提升非常显著,eventlet比没有并发的情况下,响应时间从9秒提高到0.03秒。同时eventlet/gevent提供了非阻塞的异步调用模式,非常方便。这里推荐使用线程或者伪线程,因为在响应时间类似的情况下,线程和伪线程消耗的资源更少。

本站部分内容由互联网用户自发贡献,该文观点仅代表作者本人,本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。

如发现本站有涉嫌抄袭侵权/违法违规等内容,请联系我们举报!一经查实,本站将立刻删除。

(0)

相关推荐

  • chatgtp官网,ChatGTP官网怎么下载

    《关于chatgtp官网的一些探索与思考》 一、chatgtp官网的基本情况 也许我们最先接触到的就是chatgtp官网的外观和布局了。 它的界面设计简洁大方,可能会给人一种很专业的感觉。 我觉得它的色彩搭配也很协调,不会让人感到刺眼或者不舒服。 在官网的首页,可能会有一些关于chatgtp的介绍和亮点展示,让用户能够快速了解到这个产品的大致情况。 比如,可…

    用户投稿 2025-03-21
  • 个人创新创业思维训练(有效增强创业思维的办法)

    上一期我们讲到了创业者前期必做的事情,本期老秦带你了解创业者需要具备哪些思维,如何提升思维力。思维是人类所具有的高级认识活动,是对新输入信息与脑内储存知识经验进行一系列复杂的心智操作过程 。人类把思维分为感性具象思维/抽象逻辑思维/理性具象思维。简单来说就是感官/判断力/逻辑性。思维力的强弱决定着人能力的大小,思维力可以随着后天的学习继续增加,并不是一成不变…

    用户投稿 2021-11-03
  • 个人创业有什么好项目(有什么小型创业项目)

    由于疫情尚未彻底消除,目前的创业环境相对比较恶劣。在这种情况下,想要创业赚大钱需要更加谨慎,切不可盲目投资,否则一旦创业失败,很可能很多年都翻不过身来。   下面为大家总结十个未来两年适合普通大众的创业项目。虽然这些项目技术含量并不高,但是只要把项目做好,仍然能够赚很多钱。   第一 短视频电商     现在是短视频十分火爆的时代,各大短视频平台吸…

    2021-10-29 用户投稿
  • 时尚购物网服装服饰搭配技巧

    文:洛薇 Hi,我是洛薇,继续我们的时尚穿搭之旅,变美永远不迷路。▼ 如何玩转华丽风的量身搭配?今天,它是主角。 识对人、穿对衣。挑衣选款之毒辣审美眼光的修炼,是一项硬功夫,更是一门理论 实践的时尚应用科学。 从凭感觉上升到理性购衣,其实只有两步之遥:长相选风格和身材选版型。 那么,选对衣服之后呢?如何随心所欲地穿着,则由搭配功底说了算。或者,即使选错衣服,…

    用户投稿 2023-05-22
  • 幽城幻剑录,幽城幻剑录完美结局流程攻略

    早年玩盗版游戏的人,几乎都被游戏折磨过吧,毕竟只有正版才有说明书。 很多游戏故意设计一些难点,没有说明书很容易被卡住。当年汉堂的《寰神结》和《幽城幻剑录》就是专门整治盗版的,你甚至连关闭页面都找不到。 而《金庸群侠传》这边也是如此,剧情环环相扣,场景相当复杂。没有地图、说明书和物品说明,玩家几乎是举步维艰。但偏偏早年的玩家们很有耐心,硬是靠自己的毅力将游戏内…

    2023-07-04 用户投稿
  • 大连是哪个省的(沈阳大连是哪个省的)

    最近有个全球性的榜单, 听上去就高大上的样子 全球城市500强 咱辽宁有两个城市榜上有名 沈阳,大连 值得拥有姓名! 2021年12月30日,由全球城市实验室(Global City Lab)编制的《全球城市500强》分析报告发布。 中国共有40个城市上榜,其中有两个城市来自辽宁省。大连市与沈阳市分别以691.54亿美元与429.16亿美元的城市品牌价值位列…

    2022-05-05 用户投稿
  • 类似lol的游戏,电子竞技文推荐

    本期分享4本电子竞技类游戏文,各种爽点与梗都融合得很自然,值得收藏 一、《灵笼:休闲玩家有什么坏心眼》 作者:果味喵 短评:以灵笼世界观为蓝本的虚拟网游文,背景为废土重建,竞技+网游,还有十分丰富的世界剧情,细节与情感处理得挺赞的。温润腹黑又有风度的高智商猪脚,在网游文中真的少见。退役战神回到家发现妹妹……主角算是第一公会会长付出大代价推的副本神器被讨厌主角…

    2023-07-05 用户投稿
  • 美白去斑产品,真正有效果的祛斑产品

    在改善皮肤色调这条道路上,美白精华是一定不能缺席的。对于护肤小白、学生党来说,相比于贵价大牌精华、功效成分猛药型产品,这类人群更适合从低功效型精华开始,慢慢尝试逐渐加量。 对于学生党而言,产品性价比是非常重要的。毕竟为了入手护肤品,造成钱包颤抖并不是一个明智的选择。 今天这篇内容,考拉整理出了四款价格亲民、功效口碑不错的美白精华产品,款款都是实力派,效果不比…

    2023-07-01 用户投稿
  • 苹果6s手机价格,iphone 6s现在多少钱一台

    苹果手机是换代最快的手机了,每当有新款出现的时候,旧版的就一定会降价,果不其然在传出苹果7之后,苹果6s也开始降价了,苹果7没多久又传出苹果8的消息,这也就是说现在的苹果7也开始降价了,那苹果6s现在价格是多少呢?苹果6s现在多少钱一台? 苹果iPhone6s搭配iOS9系统,与旧系统最明显的区别,就是体现在新增了一页能展开的Siri建议,苹果iPhone6…

    2023-06-02
  • 显示器对比度(显示屏的对比度多少合适)

    由于早期的液晶显示器因为画面显示不理想,所以对不同的显示特性做了比较具体的规范和量标准,其中画面的对比度是指画面中区域最亮和最暗的亮度对比,这也作为显示器的有效数据被广大用户所看重。 购买之时需谨慎 如何一秒识破对比度骗局? 目前,我们所熟知的对比度分为两种,一种是静态对比度,一种是动态对比度。相比较动态对比度,静态对比度的数值更具参考价值。但自从商家们开始…

    2023-07-06 用户投稿
  • 古茗奶茶加盟费(古茗奶茶加盟费用明细表)

    “想成为喜茶的加盟商,首先需要一次性缴纳至少41.3万元,还需要提供100万的现金或可变现资产流水,预计门店毛利率为60%。”11月4日,茶饮品牌喜茶在其微信公众号发布事业合伙细则,明确了其加盟的具体条件。 11月3日,对于市场传闻喜茶招聘加盟业务相关职位的消息,喜茶方面曾回应红星资本局,将在非一线城市以合适的店型展开事业合伙业务,在品牌、产品、品控、食安、…

    2023-05-31
  • 酸菜鱼是什么地方的菜(酸菜鱼是哪里特色菜)

    来源: 红餐网 作者:林怀青 这十年,酸菜鱼的热度基本就只臣服于小龙虾了。如果说小龙虾的发展是外来物种的走红,那么酸菜鱼的进化就该是“最本土”的进阶历史,它是中国八大菜系之川菜的经典菜式,曾经与水煮鱼、毛血旺并排“川菜三霸”,而这些年来却单飞出自己的一片新天地。 01“新派川菜”带火酸菜鱼 大约十年前,酸菜鱼是存在于正餐餐桌上的一道家常菜,不仅是川渝地区的餐…

    2022-05-06 用户投稿