使用Python的Twisted框架编写非阻塞程序的代码示例
先来看一段代码：
#~*~Twisted-APythontale~*~ fromtimeimportsleep #Hello,I'madeveloperandImainlysetupWordpress. definstall_wordpress(customer): #OurhostingcompanyThreadsLtd.isbad.Istartinstallationand... print"Startinstallationfor",customer #...thenwaittilltheinstallationfinishessuccessfully.Itis #boringandI'mspendingmostofmytimewaitingwhileconsuming #resources(memoryandsomeCPUcycles).It'sbecausetheprocess #is*blocking*. sleep(3) print"Alldonefor",customer #Idothisalldaylongforourcustomers defdeveloper_day(customers): forcustomerincustomers: install_wordpress(customer) developer_day(["Bill","Elon","Steve","Mark"])
运行一下，结果如下所示：
$./deferreds.py1
------Runningexample1------ StartinstallationforBill AlldoneforBill Startinstallation ... *Elapsedtime:12.03seconds
这是一段顺序执行的代码。四个消费者，为一个人安装需要3秒的时间，那么四个人就是12秒。这样处理不是很令人满意，所以看一下第二个使用了线程的例子：
importthreading #Thecompanygrew.WenowhavemanycustomersandIcan'thandlethe #workload.Wearenow5developersdoingexactlythesamething. defdevelopers_day(customers): #Butwenowhavetosynchronize...a.k.a.bureaucracy lock=threading.Lock() # defdev_day(id): print"Goodmorningfromdeveloper",id #Yuck-Ihatelocks... lock.acquire() whilecustomers: customer=customers.pop(0) lock.release() #MyPythonislessreadable install_wordpress(customer) lock.acquire() lock.release() print"Byefromdeveloper",id #Wegotoworkinthemorning devs=[threading.Thread(target=dev_day,args=(i,))foriinrange(5)] [dev.start()fordevindevs] #Weleavefortheevening [dev.join()fordevindevs] #Wenowgetmoredoneinthesametimebutourdevprocessgotmore #complex.Aswegrewwespendmoretimemanagingqueuesthandoingdev #work.Weevenhadoccasionaldeadlockswhenprocessesgotextremely #complex.Thefactisthatwearestillmostlypressingbuttonsand #waitingbutnowwealsospendsometimeinmeetings. developers_day(["Customer%d"%iforiinxrange(15)])
运行一下：
$./deferreds.py2
------Runningexample2------ Goodmorningfromdeveloper0Goodmorningfromdeveloper 1StartinstallationforGoodmorningfromdeveloper2 Goodmorningfromdeveloper3Customer0 ... fromdeveloperCustomer133Byefromdeveloper2 *Elapsedtime:9.02seconds
这次是一段并行执行的代码，使用了5个工作线程。15个消费者每个花费3s意味着总共45s的时间，不过用了5个线程并行执行总共只花费了9s的时间。这段代码有点复杂，很大一部分代码是用于管理并发，而不是专注于算法或者业务逻辑。另外，程序的输出结果看起来也很混杂，可读性也天津市。即使是简单的多线程的代码同样也难以写得很好，所以我们转为使用Twisted：
#Foryearswethoughtthiswasalltherewas...Wekepthiringmore #developers,moremanagersandbuyingservers.Weweretryingharder #optimisingprocessesandfire-fightingwhilegettingmediocre #performanceinreturn.Tillluckilyonedayourhosting #companydecidedtoincreasetheirfeesandwedecidedto #switchtoTwistedLtd.! fromtwisted.internetimportreactor fromtwisted.internetimportdefer fromtwisted.internetimporttask #Twistedhasaslightlydifferentapproach defschedule_install(customer): #TheyarecallingusbackwhenaWordpressinstallationcompletes. #TheyconnectedthecallerrecognitionsystemwithourCRMand #weknowexactlywhatacallisaboutandwhathastobedonenext. # #Wenowdesignprocessesofwhathastohappenoncertainevents. defschedule_install_wordpress(): defon_done(): print"Callback:Finishedinstallationfor",customer print"Scheduling:Installationfor",customer returntask.deferLater(reactor,3,on_done) # defall_done(_): print"Alldonefor",customer # #Foreachcustomer,wescheduletheseprocessesontheCRM #andthat #isallourchief-Twisteddeveloperhastodo d=schedule_install_wordpress() d.addCallback(all_done) # returnd #Yes,wedon'tneedmanydevelopersanymoreoranysynchronization. #~~Super-poweredTwisteddeveloper~~ deftwisted_developer_day(customers): print"GoodmorningfromTwisteddeveloper" # #Here'swhathastobedonetoday work=[schedule_install(customer)forcustomerincustomers] #Turnoffthelightswhendone join=defer.DeferredList(work) join.addCallback(lambda_:reactor.stop()) # print"ByefromTwisteddeveloper!" #Evenhisdayisparticularlyshort! twisted_developer_day(["Customer%d"%iforiinxrange(15)]) #Reactor,oursecretaryusestheCRMandfollows-uponevents! reactor.run()
运行结果：
------Runningexample3------ GoodmorningfromTwisteddeveloper Scheduling:InstallationforCustomer0 .... Scheduling:InstallationforCustomer14 ByefromTwisteddeveloper! Callback:FinishedinstallationforCustomer0 AlldoneforCustomer0 Callback:FinishedinstallationforCustomer1 AlldoneforCustomer1 ... AlldoneforCustomer14 *Elapsedtime:3.18seconds
这次我们得到了完美的执行代码和可读性强的输出结果，并且没有使用线程。我们并行地处理了15个消费者，也就是说，本来需要45s的执行时间在3s之内就已经完成。这个窍门就是我们把所有的阻塞的对sleep()的调用都换成了Twisted中对等的task.deferLater()和回调函数。由于现在处理的操作在其他地方进行，我们就可以毫不费力地同时服务于15个消费者。
前面提到处理的操作发生在其他的某个地方。现在来解释一下，算术运算仍然发生在CPU内，但是现在的CPU处理速度相比磁盘和网络操作来说非常快。所以给CPU提供数据或者从CPU向内存或另一个CPU发送数据花费了大多数时间。我们使用了非阻塞的操作节省了这方面的时间，例如，task.deferLater()使用了回调函数，当数据已经传输完成的时候会被激活。
另一个很重要的一点是输出中的GoodmorningfromTwisteddeveloper和ByefromTwisteddeveloper!信息。在代码开始执行时就已经打印出了这两条信息。如果代码如此早地执行到了这个地方，那么我们的应用真正开始运行是在什么时候呢？答案是，对于一个Twisted应用（包括Scrapy）来说是在reactor.run()里运行的。在调用这个方法之前，必须把应用中可能用到的每个Deferred链准备就绪，然后reactor.run()方法会监视并激活回调函数。
注意，reactor的主要一条规则就是，你可以执行任何操作，只要它足够快并且是非阻塞的。
现在好了，代码中没有那么用于管理多线程的部分了，不过这些回调函数看起来还是有些杂乱。可以修改成这样：
#Twistedgaveusutilitiesthatmakeourcodewaymorereadable! @defer.inlineCallbacks definline_install(customer): print"Scheduling:Installationfor",customer yieldtask.deferLater(reactor,3,lambda:None) print"Callback:Finishedinstallationfor",customer print"Alldonefor",customer deftwisted_developer_day(customers): ...sameaspreviouslybutusinginline_install()insteadofschedule_install() twisted_developer_day(["Customer%d"%iforiinxrange(15)]) reactor.run()
运行的结果和前一个例子相同。这段代码的作用和上一个例子是一样的，但是看起来更加简洁明了。inlineCallbacks生成器可以使用一些一些Python的机制来使得inline_install()函数暂停或者恢复执行。inline_install()函数变成了一个Deferred对象并且并行地为每个消费者运行。每次yield的时候，运行就会中止在当前的inline_install()实例上，直到yield的Deferred对象完成后再恢复运行。
现在唯一的问题是，如果我们不止有15个消费者，而是有，比如10000个消费者时又该怎样？这段代码会同时开始10000个同时执行的序列（比如HTTP请求、数据库的写操作等等）。这样做可能没什么问题，但也可能会产生各种失败。在有巨大并发请求的应用中，例如Scrapy，我们经常需要把并发的数量限制到一个可以接受的程度上。在下面的一个例子中，我们使用task.Cooperator()来完成这样的功能。Scrapy在它的ItemPipeline中也使用了相同的机制来限制并发的数目（即CONCURRENT_ITEMS设置）：
@defer.inlineCallbacks definline_install(customer): ...sameasabove #Thenew"problem"isthatwehavetomanageallthisconcurrencyto #avoidcausingproblemstoothers,butthisisaniceproblemtohave. deftwisted_developer_day(customers): print"GoodmorningfromTwisteddeveloper" work=(inline_install(customer)forcustomerincustomers) # #WeusetheCooperatormechanismtomakethesecretarynot #servicemorethan5customerssimultaneously. coop=task.Cooperator() join=defer.DeferredList([coop.coiterate(work)foriinxrange(5)]) # join.addCallback(lambda_:reactor.stop()) print"ByefromTwisteddeveloper!" twisted_developer_day(["Customer%d"%iforiinxrange(15)]) reactor.run() #Wearenowmoreleanthanever,ourcustomershappy,ourhosting #billsridiculouslylowandourperformancestellar. #~*~THEEND~*~
运行结果：
$./deferreds.py5 ------Runningexample5------ GoodmorningfromTwisteddeveloper ByefromTwisteddeveloper! Scheduling:InstallationforCustomer0 ... Callback:FinishedinstallationforCustomer4 AlldoneforCustomer4 Scheduling:InstallationforCustomer5 ... Callback:FinishedinstallationforCustomer14 AlldoneforCustomer14 *Elapsedtime:9.19seconds
从上面的输出中可以看到，程序运行时好像有5个处理消费者的槽。除非一个槽空出来，否则不会开始处理下一个消费者的请求。在本例中，处理时间都是3秒，所以看起来像是5个一批次地处理一样。最后得到的性能跟使用线程是一样的，但是这次只有一个线程，代码也更加简洁更容易写出正确的代码。
PS：deferToThread使同步函数实现非阻塞
wisted的defer.Deferred(fromtwisted.internetimportdefer)可以返回一个deferred对象.
注:deferToThread使用线程实现的,不推荐过多使用
***把同步函数变为异步(返回一个Deferred)***
twisted的deferToThread(fromtwisted.internet.threadsimportdeferToThread)也返回一个deferred对象,不过回调函数在另一个线程处理,主要用于数据库/文件读取操作
..
#代码片段
defdataReceived(self,data):
now=int(time.time())
forftype,datainself.fpcodec.feed(data):
ifftype=='oob':
self.msg('OOB:',repr(data))
elifftype==0x81:#对服务器请求的心跳应答(这个是解析防疲劳驾驶仪,发给gps上位机的,然后上位机发给服务器的)
self.msg('FP.PONG:',repr(data))
else:
self.msg('TODO:',(ftype,data))
d=deferToThread(self.redis.zadd,"beier:fpstat:fps",now,self.devid)
d.addCallback(self._doResult,extra)
下面这儿完整的例子可以给大家参考一下
#-*-coding:utf-8-*- fromtwisted.internetimportdefer,reactor fromtwisted.internet.threadsimportdeferToThread importfunctools importtime #耗时操作这是一个同步阻塞函数 defmySleep(timeout): time.sleep(timeout) #返回值相当于加进了callback里 return3 defsay(result): print"耗时操作结束了,并把它返回的结果给我了",result #用functools.partial包装一下,传递参数进去 cb=functools.partial(mySleep,3) d=deferToThread(cb) d.addCallback(say) print"你还没有结束我就执行了,哈哈" reactor.run()