最近项目中由于在 python3 中使用 tornado,之前也有用过，是在 python2 中，由于对于协程理解不是很透彻，只是套用官方文档中的写法，最近比较细致的看了下协程的用法，也将 tornado 在 python3 中异步的实践了一下。
要理解协程，先要理解异步，要理解异步，先要理解同步，与同步相关的概念又有阻塞与非阻塞，下面一一做简单介绍。
阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的。
常见的阻塞形式有：网络 I/O 阻塞、磁盘 I/O 阻塞、用户输入阻塞等。阻塞是无处不在的，包括 CPU 切换上下文时，所有的进程都无法真正干事情，它们也会被阻塞。如果是多核 CPU 则正在执行上下文切换操作的核不可被利用。
程序在等待某操作过程中，自身不被阻塞，可以继续运行干别的事情，则称该程序在该操作上是非阻塞的。
非阻塞并不是在任何程序级别、任何情况下都可以存在的。
仅当程序封装的级别可以囊括独立的子程序单元时，它才可能存在非阻塞状态。
非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞的。
不同程序单元为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，称这些程序单元是同步执行的。
例如购物系统中更新商品库存，需要用 “行锁” 作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。
简言之，同步意味着有序。
为完成某个任务，不同程序单元之间过程中无需通信协调，也能完成任务的方式，不相关的程序单元之间可以是异步的。
例如，爬虫下载网页。调度程序调用下载程序后，即可调度其他任务，而无需与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无需相互通知协调。这些异步操作的完成时刻并不确定。
简言之，异步意味着无序。
这个概念让我想起了上学时学过的一篇文章，讲统筹安排的，比如你现在要烧水，做饭，洗衣服三件事，如果同步的进行，先烧水，在水烧开的过程中你什么都不做就等着它烧开，然后水烧开以后你再接着做饭，饭做熟的过程中你也是什么都不干，就干等着，饭做熟后再去将洗衣服放入洗衣机中去洗，之后又是干等着。如果单做一件事的时间是烧水 10 分别，做饭 30 分钟，洗衣服 20 分钟，那么完成这三件事总共需要 60 分钟。
如果将这三件事异步的去进行，我先将水烧上，然后再将衣服放到洗衣机里，然后去做饭，这三件事同时进行，当水烧开的时候给我一个信号，这里就是水壶会响，我听到响声以后我会中止做饭这件事情去处理烧开的水，比如把它倒到保温瓶中，衣服洗完以后洗衣机也会给我一个信号，那么我就会将衣服拿出来晾晒。这样处理完三件事总共的时间就由三件事情中最长的时间决定，这里就是 30 分钟，其实异步的处理就是最大程度的发挥 cpu 的处理能力，让其在同一时间内做更多的事情。
上面的过程用代码来实现大概是这个样子
#coding:gbk
import time
import threading,multiprocessing
def water():
print("开始烧水了")
start = time.time()
time.sleep(2)
end = time.time()
return ("水开了", end - start)
def wash():
print("开始洗衣服了")
start = time.time()
time.sleep(4)
end = time.time()
return ("衣服洗完了", end - start)
def cook():
print("开始做饭了")
start = time.time()
time.sleep(10)
end = time.time()
return ("饭熟了",end - start)
def alarm(info):
print("%s 一共用了%s 时间"%(info[0],info[1]))
class DoWork(threading.Thread):
def __init__(self, func, callback,*params, **paramMap):
threading.Thread.__init__(self)
self.func = func
self.params = params
self.paramMap = paramMap
self.rst = None
self.finished = False
self.isDaemon = True
self.isDaemon = False
self.callback = callback
def run(self):
self.rst = self.func(*self.params, **self.paramMap)
self.callback(self.rst)
self.finished = True
if __name__ == '__main__':
task1 = DoWork(water,alarm)
task2 = DoWork(wash,alarm)
task3 = DoWork(cook,alarm)
start = time.time()
task1.start()
task2.start()
task3.start()
task1.join()
task2.join()
task3.join()
end = time.time()
print("一共用了%s时间"%(end-start))
执行结果如下
开始烧水了
开始洗衣服了
开始做饭了
水开了 一共用了2.0003998279571533 时间
衣服洗完了 一共用了4.00029993057251 时间
饭熟了 一共用了10.000999689102173 时间
一共用了10.001499891281128时间
yield 语法以上是用了多线程的方式来达到异步的效果，但是并没有用到协程，协程在 python2 就有，现在来看看在 python2 中通过yield语法。以下方法是在 python2.6 中执行.
要了解 yield 语法，先要了解一个概念： Generator 『生成器』，关于 generator 的概念可以参考廖雪峰的教程，写的很好，生成器
如果一个函数定义中包含 yield 关键字，那么这个函数就不再是一个普通函数，而是一个 generator
def h():
print('I am yangyanxing')
yield 5
if __name__ == '__main__':
c = h()
运行该脚本以后程序并没有任何输出，因为它有 yield 表达式，因此，我们通过 next() 语句让它执行。next() 语句将恢复 Generator 执行，并直到下一个 yield 表达式处。比如：
def h():
print('I am yangyanxing')
yield 5
print("I am fjy")
if __name__ == '__main__':
c = h()
c.next()
调用 c.nect() 以后，函数开始执行，这时先打印 "I am yangyanxing", 之后遇到 yield 关键字，此时函数又被中断，脚本执行结束，程序只打印了一行 "I am yangyanxing",如果想要打印出 I am fjy 呢，以时需要再调用一次 c.next(), 当再次调用 c.next() 时，函数从之前的 yield 处开始执行，由于函数在之后没有 yield 了，所以程序会抛一个 StopIteration 异常。
与 next() 函数相关的还有一个 send() 函数，next 函数传递的是 None ， send 函数可以传递对应的值。其实 next() 和 send() 在一定意义上作用是相似的，区别是 send() 可以传递 yield 表达式的值进去，而 next() 不能传递特定的值，只能传递 None 进去。因此，我们可以看做
c.next() 和 c.send(None) 作用是一样的。看如下的代码。
def h():
print('I am yangyanxing')
m = yield 5
print(m)
print("I am fjy")
yield 6
print("They love too much")
if __name__ == '__main__':
c = h()
c.next()
c.send("hahaha")
函数的输出为
I am yangyanxing
hahaha
I am fjy
这代码解析，首先通过 c = h() 定义了一个 generator ,然后调用 c.next() 启动这个生成器，生成器先打印出 I am yangyanxing 然后遇到 m = yield 5 这行代码，后停止，之后再调用 c.send("hahaha") ，这时候 m 的值就是 hahaha, 然后再打印出 m ,之后再打印出 I am fjy,之后又遇到了 yield 关键字，程序又中止了，整个脚本执行结束,需要提醒的是，第一次调用时，请使用 next() 语句或是 send(None)，不能使用 send 发送一个非 None 的值，否则会出错的，因为没有 yield 语句来接收这个值。
那么 next() 与 send() 函数的返回值是什么呢？ 注意到上面函数中的 yield 之后是一个 5 了吗？其实这就是调用 netx 或者 send 以后得到的返回值.
def h():
print('I am yangyanxing')
m = yield 5
print(m)
print("I am fjy")
yield 6
if __name__ == '__main__':
c = h()
print(c.next())
print(c.send("hahaha"))
得到的输出为
I am yangyanxing
5
hahaha
I am fjy
6
首先看以下的代码，以下是在 python2 中编写
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import time
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
class IndexHandler(tornado.web.RequestHandler):
def get(self):
query = self.get_argument('q')
time.sleep(5)
self.write("hello %s"%query)
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[(r"/", IndexHandler)])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
我分别用浏览器和和用脚本对 http://127.0.0.1:8000/?q=yangyanxing 该 url 进行访问，脚本如下
import requests,time
def geturl(url):
s = time.time()
r = requests.get(url)
e = time.time()
print(int(e-s))
getrul(r"http://127.0.0.1:8000/?q=yangyanxing")
服务端显示
[I 190114 00:03:46 web:2162] 304 GET /?q=yangyanxing (127.0.0.1) 5000.97ms
[I 190114 00:03:51 web:2162] 200 GET /?q=yangyanxing (127.0.0.1) 5006.78ms
脚本打印为 7 或者 8
在同步应用中，由于同时只能提供一个请求。所以，如果一个路由中有一个比较耗时的操作，如代码中的 time.sleep(5) 那么意味着如果同时有两个请求，那么第二个请求只能等待服务器处理完第一个请求之后才能处理第二个请求，也就中处理两个请求，最短要 5 秒，最长要 10 秒,这还只是 2 个，如果有 10 个那就是要 50 秒处理完所有的请求，这个效率是无法接受的，服务端可否同时处理 10 个请求，就像文章一开始提到的同时做三件事情，在处理一个耗时的操作时，不是干等着这件事情处理完，而是去做别的事情，当那件事情结束以后，再通过调用回调函数来通知调用者。
#coding:utf-8
import time
import requests
def geturl(url):
start = time.time()
r = requests.get(url)
end = time.time()
print("用了%s时间"%(end-start))
return r.status_code
if __name__ == '__main__':
start = time.time()
for i in range(10):
geturl(r"https://www.baidu.com")
end = time.time()
print("all done,use %s time"%(end-start))
执行结果如下：
用了0.2974998950958252时间
用了0.0494999885559082时间
用了0.0410001277923584时间
用了0.029999971389770508时间
用了0.18199992179870605时间
用了0.02700018882751465时间
用了0.022499799728393555时间
用了0.020000219345092773时间
用了0.020999908447265625时间
用了0.019999980926513672时间
all done,use 0.7105000019073486 time
一共用了 0.7 秒，百度的反应可能太快了，我们准备一个本地的环境来模拟慢返回。这里我先使用 tornado 的异步协程处理，之后再详细说明该处的用法。
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import asyncio
import urllib
import json
import datetime
import time
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
class IndexHandler(tornado.web.RequestHandler):
async def get(self):
query = self.get_argument('q')
await asyncio.sleep(5)
self.writeres("hello %s"%query)
self.finish()
def writeres(self,returnstr):
self.write(returnstr)
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[(r"/", IndexHandler)])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
请求代码改为三次，只是为了说明问题
def geturl(url):
start = time.time()
r = requests.get(url)
end = time.time()
print("用了%s时间"%(end-start))
return r.status_code
if __name__ == '__main__':
start = time.time()
for i in range(3):
geturl(r"http://127.0.0.1:8000/?q=yangyanxing")
end = time.time()
print("all done,use %s time"%(end-start))
结果：
用了5.009501695632935时间
用了5.012002229690552时间
用了5.012502193450928时间
all done,use 15.035006284713745 time
可以看到，总是时间是 15 秒，同步对一个 url 发请求，在没有做异步处理的时候时间是累积的。
接下来说本篇的重点，协程.
在一个普通的函数前面加上 async 关键字，此时该函数会返回一个 coroutine 对象，函数里也不会立刻执行.
async def getUrlByCor(url):
start = time.time()
r = requests.get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r.status_code
if __name__ == '__main__':
s = getUrlByCor(r'http://127.0.0.1:8000/?q=yangyanxing')
print(s)
运行结果：
此处的 s 是一个 coroutine 对象，那么怎么才能执行函数里面的方法呢？ 将这个协程对象放到事件循环 event_loop 中执行
async def getUrlByCor(url):
start = time.time()
r = requests.get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r.status_code
if __name__ == '__main__':
s = getUrlByCor(r"http://127.0.0.1:8000/?q=yangyanxing")
loop = asyncio.get_event_loop()
loop.run_until_complete(s)
执行结果：
用了 5.009500026702881 时间
如果同时发三个请求呢？
这里要对协程做一个封装，将其封装成一个 task 对象
async def getUrlByCor(url):
start = time.time()
r = requests.get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r.text
if __name__ == '__main__':
s = time.time()
tasks = [asyncio.ensure_future(getUrlByCor(r"http://127.0.0.1:8000/?q=yangyanxing")) for i in range(3)]
print(len(tasks),tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
e = time.time()
print("use %s time"%(e-s))
for task in tasks:
print('Task Result:', task.result())
结果如下：
3 [<Task pending coro=<getUrlByCor() running at E:/Github/asyncTorMysql/asynctest.py:69>>, <Task pending coro=<getUrlByCor() running at E:/Github/asyncTorMysql/asynctest.py:69>>, <Task pending coro=<getUrlByCor() running at E:/Github/asyncTorMysql/asynctest.py:69>>]
用了5.008501291275024时间
用了5.012002229690552时间
用了5.012002229690552时间
use 15.03450632095337 time
Task Result: hello yangyanxing
Task Result: hello yangyanxing
Task Result: hello yangyanxing
总的时间还是 15 秒，并没有实现异步呢，还是同步的依次执行请求。
其实，要实现异步处理，我们得先要有挂起的操作，当一个任务需要等待 IO 结果的时候，可以挂起当前任务，转而去执行其他任务，这样我们才能充分利用好资源，上面方法都是一本正经的串行走下来，连个挂起都没有，怎么可能实现异步？
上面的函数存在耗时的操作就是r = requests.get(url) 那么将它写成挂起的呢？
r = await requests.get(url)
async def getUrlByCor(url):
start = time.time()
r = await requests.get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r.text
if __name__ == '__main__':
s = time.time()
tasks = [asyncio.ensure_future(getUrlByCor(r"http://127.0.0.1:8000/?q=yangyanxing")) for i in range(3)]
print(len(tasks),tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
e = time.time()
print("use %s time"%(e-s))
for task in tasks:
print('Task Result:', task.result())
不出意外的报错了
Task exception was never retrieved
future: exception=TypeError("object Response can't be used in 'await' expression",)>
Traceback (most recent call last):
File "C:\Python35\lib\asyncio\tasks.py", line 240, in _step
result = coro.send(None)
File "E:/Github/asyncTorMysql/asynctest.py", line 71, in getUrlByCor
r = await requests.get(url)
TypeError: object Response can't be used in 'await' expression
这个错误的意思是 requests 返回的 Response 对象不能和 await 一起使用,await 后面的对象必须是如下格式之一
reqeusts 返回的 Response 不符合上面任一条件，因此就会报上面的错误了。
既然 await 后面的对象要是 coroutine 对象 ，那么将其包装在async 后面不就可以了吗？
async def get(url):
return requests.get(url)
async def getUrlByCor(url):
start = time.time()
r = await get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r.text
if __name__ == '__main__':
s = time.time()
tasks = [asyncio.ensure_future(getUrlByCor(r"http://127.0.0.1:8000/?q=yangyanxing")) for i in range(3)]
print(len(tasks),tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
e = time.time()
print("use %s time"%(e-s))
for task in tasks:
print('Task Result:', task.result())
这次不报错了，但是依然没有异步的执行
用了5.0090014934539795时间
用了5.011002063751221时间
用了5.011502027511597时间
use 15.03450632095337 time
Task Result: hello yangyanxing
Task Result: hello yangyanxing
Task Result: hello yangyanxing
也就是说我们仅仅将涉及 IO 操作的代码封装到 async 修饰的方法里面是不可行的！我们必须要使用支持异步操作的请求方式才可以实现真正的异步，所以这里就需要 aiohttp 派上用场了
aiohttp 是一个支持异步请求的库，利用它和 asyncio 配合我们可以非常方便地实现异步请求操作。
async def get(url):
session = aiohttp.ClientSession()
res = await session.get(url)
result = await res.text()
await session.close()
return result
async def getUrlByCor(url):
start = time.time()
r = await get(url)
end = time.time()
print("用了%s时间" % (end - start))
return r
if __name__ == '__main__':
s = time.time()
tasks = [asyncio.ensure_future(getUrlByCor(r"http://127.0.0.1:8000/?q=yangyanxing")) for i in range(3)]
print(len(tasks),tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
e = time.time()
print("use %s time"%(e-s))
for task in tasks:
print('Task Result:', task.result())
执行结果：
用了5.006500005722046时间
用了5.006499767303467时间
用了5.006500005722046时间
use 5.008500099182129 time
Task Result: hello yangyanxing
Task Result: hello yangyanxing
Task Result: hello yangyanxing
这次终于实现了异步请求。
还记得最开始的洗衣做饭的例子吗？可以使用异步协程来实现，代码大概是这个样子
#coding:utf-8
import time
import threading,multiprocessing
import requests
import asyncio
import aiohttp
async def water():
print("开始烧水了")
start = time.time()
await asyncio.sleep(2) # 这里使用asyncio.sleep 异步休眠
end = time.time()
return ("水开了", end - start)
async def wash():
print("开始洗衣服了")
start = time.time()
await asyncio.sleep(4)
end = time.time()
return ("衣服洗完了", end - start)
async def cook():
print("开始做饭了")
start = time.time()
await asyncio.sleep(10)
end = time.time()
return ("饭熟了",end - start)
def alarm(task):
print("%s 一共用了%s 时间"%(task.result()[0],task.result()[1]))
if __name__ == '__main__':
tasks = []
# 初始化task
task_water = asyncio.ensure_future(water())
task_wash = asyncio.ensure_future(wash())
task_cook = asyncio.ensure_future(cook())
tasks.extend([task_water,task_wash,task_cook])
# 将这三个task 绑定回调函数
task_water.add_done_callback(alarm)
task_wash.add_done_callback(alarm)
task_cook.add_done_callback(alarm)
# 将task放入事件循环中
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
执行结果：
开始烧水了
开始洗衣服了
开始做饭了
水开了 一共用了 2.0 时间
衣服洗完了 一共用了 4.003999948501587 时间
饭熟了 一共用了 9.994499921798706 时间
我们再来翻过头来看之前用 tornado 写的服务端同步代码
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import time
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
class IndexHandler(tornado.web.RequestHandler):
def get(self):
query = self.get_argument('q')
time.sleep(5)
self.write("hello %s"%query)
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[(r"/", IndexHandler)])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
在 IndexHandler 中的 get 方法，由于当中存在了一个比较耗时的操作，time.sleep(5) 处理完这个请求需要卡 5 秒，在卡住的这段时间，tornado 无法再完成别的请求，如果此时再发来一个 / 的请求，那么只能等待这前的请求操作结束之后再对处理新发过来的请求，如果同时有 1 万个请求发过来，可想而知，最后一个请求就等到猴年马月才能处理完呢……
解决方法是使用@tornado.web.asynchronous 和@tornado.gen.coroutine 装饰器，将耗时的操作放到线程中去执行，这里的耗时操作 time.sleep(5) 是阻塞的，所以将阻塞函数放加上 @run_on_executor 装饰器
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import tornado.autoreload
from tornado.concurrent import run_on_executor
import time
from concurrent.futures import ThreadPoolExecutor
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
class IndexHandler(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(100)
@user4nous
@user5e
def get(self):
query = self.get_argument('q')
res = yield self.sleepBlock(2,query)
self.write("hello %s"%res)
self.finish()
@run_on_executor
def sleepBlock(self,sleeptime,query):
time.sleep(sleeptime)
return query
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[(r"/", IndexHandler)],settings={"debug":True})
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
注意到在 IndexHandler 中有一行初始化 executor 的代码 executor = ThreadPoolExecutor(100) 这里的参数 100 是最大的线程数，我这里传的是 100，也就意味着同时能处理 100 个请求，当有 101 个请求的时候，前 100 个请求可以同时在 2 秒内执行，最后的那一个请求就要等之前有结束的线程以后再去执行了。
由于在 python3.5 以后引入了 asyncio这个标准库，很多异步的操作可以用这个库来操作
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import tornado.autoreload
from tornado.concurrent import run_on_executor
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
class IndexHandler(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(100)
@user7nous
@user8e
def get(self):
query = self.get_argument('q')
res = yield self.sleepBlock(2,query)
self.write("hello %s"%res)
self.finish()
@run_on_executor
def sleepBlock(self,sleeptime,query):
time.sleep(sleeptime)
return query
class asyncIndexHandler(tornado.web.RequestHandler):
async def get(self):
query = self.get_argument('q')
res = await self.sleepBlock(2,query)
self.write("hello %s"%res)
self.finish()
async def sleepBlock(self,sleeptime,query):
await asyncio.sleep(sleeptime) # 使用异步的sleep方法
return query
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[
(r"/", IndexHandler),
(r"/async",asyncIndexHandler)
],settings={"debug":True})
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
IndexHandler 中的 get 方法使用了async 与await 关键字来达到异步的处理请求，这里的asyncio.sleep(5) 是异步的暂停 5 秒，如果此处的方法涉及到无法使用异步请求的库该怎么处理，比如说我就想使用time.sleep(5) 则需要在线程池中运行，就像上面的/ 路由里使用 @run_on_executor 中执行。
异步操作涉及的知识点比较多，不同版本的 python 对于异步的处理也不一样，有些东西如 yield 理解起来比较费劲，需要多在项目中实践，tornado 这个框架的设计初衷也是异步网络库，过使用非阻塞网络 I/O, Tornado 可以支持上万级的连接，所以要使用过程中要多多考虑异步非阻塞的编码。
参考文章
理解 Python 协程:从 yield/send 到 yield from 再到 async/await