先来看一段代码:

# ~*~ Twisted - A Python tale ~*~from time import sleep# Hello, I'm a developer and I mainly setup Wordpress.
def install_wordpress(customer): # Our hosting company Threads Ltd. is bad. I start installation and... print "Start installation for", customer # ...then wait till the installation finishes successfully. It is # boring and I'm spending most of my time waiting while consuming # resources (memory and some CPU cycles). It's because the process # is *blocking*. sleep(3) print "All done for", customer # I do this all day long for our customers def developer_day(customers): for customer in customers: install_wordpress(customer) developer_day(["Bill", "Elon", "Steve", "Mark"])
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

运行一下,结果如下所示:

$ ./deferreds.py 1
------ Running example 1 ------
Start installation for Bill
All done for Bill
Start installation
... * Elapsed time: 12.03 seconds
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

这是一段顺序执行的代码。四个消费者,为一个人安装需要3秒的时间,那么四个人就是12秒。这样处理不是很令人满意,所以看一下第二个使用了线程的例子:

import threading# The company grew. We now have many customers and I can't handle the
# workload. We are now 5 developers doing exactly the same thing.
def developers_day(customers): # But we now have to synchronize... a.k.a. bureaucracy lock = threading.Lock() # def dev_day(id): print "Goodmorning from developer", id # Yuck - I hate locks... lock.acquire() while customers: customer = customers.pop(0) lock.release() # My Python is less readable install_wordpress(customer) lock.acquire() lock.release() print "Bye from developer", id # We go to work in the morning devs = [threading.Thread(target=dev_day, args=(i,)) for i in range(5)] [dev.start() for dev in devs] # We leave for the evening [dev.join() for dev in devs] # We now get more done in the same time but our dev process got more # complex. As we grew we spend more time managing queues than doing dev # work. We even had occasional deadlocks when processes got extremely # complex. The fact is that we are still mostly pressing buttons and # waiting but now we also spend some time in meetings. developers_day(["Customer %d" % i for i in xrange(15)])
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32

运行一下:

$ ./deferreds.py 2
------ Running example 2 ------
Goodmorning from developer 0Goodmorning from developer
1Start installation forGoodmorning from developer 2
Goodmorning from developer 3Customer 0
... from developerCustomer 13 3Bye from developer 2 * Elapsed time: 9.02 seconds
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

这次是一段并行执行的代码,使用了5个工作线程。15个消费者每个花费3s意味着总共45s的时间,不过用了5个线程并行执行总共只花费了9s的时间。这段代码有点复杂,很大一部分代码是用于管理并发,而不是专注于算法或者业务逻辑。另外,程序的输出结果看起来也很混杂,可读性也天津市。即使是简单的多线程的代码同样也难以写得很好,所以我们转为使用Twisted:

# For years we thought this was all there was... We kept hiring more
# developers, more managers and buying servers. We were trying harder
# optimising processes and fire-fighting while getting mediocre
# performance in return. Till luckily one day our hosting
# company decided to increase their fees and we decided to # switch to Twisted Ltd.! from twisted.internet import reactor from twisted.internet import defer from twisted.internet import task # Twisted has a slightly different approach def schedule_install(customer): # They are calling us back when a Wordpress installation completes. # They connected the caller recognition system with our CRM and # we know exactly what a call is about and what has to be done next. # # We now design processes of what has to happen on certain events. def schedule_install_wordpress(): def on_done(): print "Callback: Finished installation for", customer print "Scheduling: Installation for", customer return task.deferLater(reactor, 3, on_done) # def all_done(_): print "All done for", customer # # For each customer, we schedule these processes on the CRM # and that # is all our chief-Twisted developer has to do d = schedule_install_wordpress() d.addCallback(all_done) # return d # Yes, we don't need many developers anymore or any synchronization. # ~~ Super-powered Twisted developer ~~ def twisted_developer_day(customers): print "Goodmorning from Twisted developer" # # Here's what has to be done today work = [schedule_install(customer) for customer in customers] # Turn off the lights when done join = defer.DeferredList(work) join.addCallback(lambda _: reactor.stop()) # print "Bye from Twisted developer!" # Even his day is particularly short! twisted_developer_day(["Customer %d" % i for i in xrange(15)]) # Reactor, our secretary uses the CRM and follows-up on events! reactor.run()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

运行结果:

------ Running example 3 ------
Goodmorning from Twisted developer
Scheduling: Installation for Customer 0
....
Scheduling: Installation for Customer 14 Bye from Twisted developer! Callback: Finished installation for Customer 0 All done for Customer 0 Callback: Finished installation for Customer 1 All done for Customer 1 ... All done for Customer 14 * Elapsed time: 3.18 seconds
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

这次我们得到了完美的执行代码和可读性强的输出结果,并且没有使用线程。我们并行地处理了15个消费者,也就是说,本来需要45s的执行时间在3s之内就已经完成。这个窍门就是我们把所有的阻塞的对sleep()的调用都换成了Twisted中对等的task.deferLater()和回调函数。由于现在处理的操作在其他地方进行,我们就可以毫不费力地同时服务于15个消费者。

前面提到处理的操作发生在其他的某个地方。现在来解释一下,算术运算仍然发生在CPU内,但是现在的CPU处理速度相比磁盘和网络操作来说非常快。所以给CPU提供数据或者从CPU向内存或另一个CPU发送数据花费了大多数时间。我们使用了非阻塞的操作节省了这方面的时间,例如,task.deferLater()使用了回调函数,当数据已经传输完成的时候会被激活。

另一个很重要的一点是输出中的Goodmorning from Twisted developerBye from Twisted developer!信息。在代码开始执行时就已经打印出了这两条信息。如果代码如此早地执行到了这个地方,那么我们的应用真正开始运行是在什么时候呢?答案是,对于一个Twisted应用(包括Scrapy)来说是在reactor.run()里运行的。在调用这个方法之前,必须把应用中可能用到的每个Deferred链准备就绪,然后reactor.run()方法会监视并激活回调函数。

注意,reactor的主要一条规则就是,你可以执行任何操作,只要它足够快并且是非阻塞的。

现在好了,代码中没有那么用于管理多线程的部分了,不过这些回调函数看起来还是有些杂乱。可以修改成这样:

# Twisted gave us utilities that make our code way more readable!
@defer.inlineCallbacks
def inline_install(customer): print "Scheduling: Installation for", customer yield task.deferLater(reactor, 3, lambda: None) print "Callback: Finished installation for", customer print "All done for", customer def twisted_developer_day(customers): ... same as previously but using inline_install() instead of schedule_install() twisted_developer_day(["Customer %d" % i for i in xrange(15)]) reactor.run()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

运行的结果和前一个例子相同。这段代码的作用和上一个例子是一样的,但是看起来更加简洁明了。inlineCallbacks生成器可以使用一些一些Python的机制来使得inline_install()函数暂停或者恢复执行。inline_install()函数变成了一个Deferred对象并且并行地为每个消费者运行。每次yield的时候,运行就会中止在当前的inline_install()实例上,直到yieldDeferred对象完成后再恢复运行。

现在唯一的问题是,如果我们不止有15个消费者,而是有,比如10000个消费者时又该怎样?这段代码会同时开始10000个同时执行的序列(比如HTTP请求、数据库的写操作等等)。这样做可能没什么问题,但也可能会产生各种失败。在有巨大并发请求的应用中,例如Scrapy,我们经常需要把并发的数量限制到一个可以接受的程度上。在下面的一个例子中,我们使用task.Cooperator()来完成这样的功能。Scrapy在它的Item Pipeline中也使用了相同的机制来限制并发的数目(即CONCURRENT_ITEMS设置):

@defer.inlineCallbacks
def inline_install(customer): ... same as above # The new "problem" is that we have to manage all this concurrency to # avoid causing problems to others, but this is a nice problem to have. def twisted_developer_day(customers): print "Goodmorning from Twisted developer" work = (inline_install(customer) for customer in customers) # # We use the Cooperator mechanism to make the secretary not # service more than 5 customers simultaneously. coop = task.Cooperator() join = defer.DeferredList([coop.coiterate(work) for i in xrange(5)]) # join.addCallback(lambda _: reactor.stop()) print "Bye from Twisted developer!" twisted_developer_day(["Customer %d" % i for i in xrange(15)]) reactor.run() # We are now more lean than ever, our customers happy, our hosting # bills ridiculously low and our performance stellar. # ~*~ THE END ~*~
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

运行结果:

$ ./deferreds.py 5
------ Running example 5 ------
Goodmorning from Twisted developer
Bye from Twisted developer!
Scheduling: Installation for Customer 0
... Callback: Finished installation for Customer 4 All done for Customer 4 Scheduling: Installation for Customer 5 ... Callback: Finished installation for Customer 14 All done for Customer 14 * Elapsed time: 9.19 seconds
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

从上面的输出中可以看到,程序运行时好像有5个处理消费者的槽。除非一个槽空出来,否则不会开始处理下一个消费者的请求。在本例中,处理时间都是3秒,所以看起来像是5个一批次地处理一样。最后得到的性能跟使用线程是一样的,但是这次只有一个线程,代码也更加简洁更容易写出正确的代码。

理解Twisted与非阻塞编程相关推荐

  1. Java异步非阻塞编程的几种方式

    简介: Java异步非阻塞编程的几种方式 一. 从一个同步的Http调用说起 一个很简单的业务逻辑,其他后端服务提供了一个接口,我们需要通过接口调用,获取到响应的数据. 逆地理接口:通过经纬度获取这个 ...

  2. Java中的NIO非阻塞编程

    在JDK1.4以前,Java的IO操作集中在java.io这个包中,是基于流的阻塞API.对于大多数应用来说,这样的API使用很方便,然而,一些对性能要求较高的应用,尤其是服务端应用,往往需要一个更为 ...

  3. mysql异步非阻塞方式_如何理解swoole异步非阻塞?

    传统的apache2handler或php-fpm本质上都是短生命周期(请求后释放资源)的FastCGI运行模式. 请求来了,master进程会调用worker进程来处理,处理完后释放资源. 假设你在 ...

  4. Web-Flux非阻塞编程

    普通的Controller @GetMapping("/1")public String fun1() {log.info("get str in fun1 start& ...

  5. Linux C/C++网络编程实战-陈硕-笔记20-使用非阻塞IO

    基于IO复用(非阻塞IO)实现的 netcat 使用非阻塞IO可以有效避免上述情况的发生.但非阻塞IO在编程上要比阻塞IO更难,并且在程序的维护上比较痛苦.一般使用非阻塞IO编程时建议使用一些封装好的 ...

  6. Spring WebFlux异步非阻塞式编程

    一.什么是 Spring WebFlux Spring MVC 构建于 Servlet API 之上,使用的是同步阻塞式 I/O 模型,什么是同步阻塞式 I/O 模型呢?就是说,每一个请求对应一个线程 ...

  7. 深度好文:select与非阻塞IO

    对于面向连接的socket类型(SOCK_STREAM,SOCK_SEQPACKET)在读写数据之前必须建立连接,首先服务器端socket必须在一个客户端知道的地址进行监听,也就是创建socket之后 ...

  8. Socket,非阻塞,fcntl

    随笔 - 71  文章 - 61  评论 - 32 一.fcntl 用以下方法将socket设置成为非阻塞方式 int  flags = fcntl(socket,F_GETFL,0); fcntl( ...

  9. 用Java实现非阻塞通信

    用ServerSocket和Socket来编写服务器程序和客户程序,是Java网络编程的最基本的方式.这些服务器程序或客户程序在运行过程中常常会阻塞.例如当一个线程执行ServerSocket的acc ...

最新文章

  1. Oracle the network adapter could not establish the connection
  2. SOA流程项目到底是业务流还是工作流
  3. Service Mesh 是什么,我们为什么需要它?
  4. openssl 学习之从证书中提取RSA公钥N 和 E
  5. 浅蓝色设计类网站模板
  6. java构造器_Java类加载的过程
  7. 富士康已看到芯片短缺开始缓解迹象 预计下半年会有改善
  8. Flink的定时器(EventTime和ProcessTime)
  9. UIScrollView无法滚动
  10. ssms 连接 ssis_在SSMS中手动设置SSIS包加密
  11. Spring入门第十七课
  12. 李开复、张亚勤、吴恩达…国际大咖给你讲解AI知识
  13. 如何排查即时通讯互动开发平台AnyChat的故障信息
  14. 新版 世界地图 中文版地图 国界地图 高清全彩矢量地图 CDR 2021年整理制作
  15. Iterator迭代器遍历Map集合
  16. Android 扫描二维码、制作二维码、识别图片中的二维码;简单的Zxing扫一扫功能;
  17. 【参考文献】支气管上皮细胞培养方法
  18. mysql常用知识点
  19. 微软surface屏幕测试软件,Soomal作品 - Microsoft 微软 Surface Go平板电脑屏幕测评报告 [Soomal]...
  20. 双系统安装deepin20_记录一次Windows+Deepin双系统安装及简单优化

热门文章

  1. Unity Application Block 1.0系列(7): Lifetime Managers
  2. 查看SAP CRM和C4C的UI technical信息 1
  3. 01-NLP-02-gensim中文处理案例
  4. javascript原生代码取单选框的值
  5. JavaScript——语法
  6. nginx websocket 代理
  7. ovirt 添加存储域错误提示
  8. SAP BPC最佳实践-BPC安装及配置的常见问题
  9. 计算色光叠加的RGB颜色的alpha值
  10. 服务器动态上下线监听案例