tornado.gen.coroutine-编写异步函数

摘要: 前面已经介绍了coroutine的运行机制，了解原理后，就得学习写自己的异步函数，用于支持coroutine装饰。

异步函数：

1. 返回Future

2. 必须有set_result( )或者set_exception( )调用。

这里展示一个异步socket读取的例子：

首先定义一个定时返回的服务器，来模拟耗时的操作

from tornado.tcpserver import TCPServer
from tornado import ioloop
from tornado import gen
from tornado.concurrent import Future

def sleep(duration):
f = Future()
ioloop.IOLoop.current().call_later(duration, lambda: f.set_result(None))
return f

def handle_excep(future):
if future.exception() is not None:
print future.exc_info()

class EchoServer(TCPServer):
def handle_stream(self, stream, address):
f = self._handle_stream(stream, address)
f.add_done_callback(handle_excep)

@gen.coroutine
def _handle_stream(self, stream, address):
yield data = yield stream.read_until('\n')
yield sleep(2)
yield stream.write(data)
stream.close()

server = EchoServer()
server.listen(8888)
ioloop.IOLoop.instance().start()

这里使用sleep( )函数，是模拟耗时操作。

sleep函数在tornado 4.1在gen模块中有定义，可以直接使用gen.sleep。

HTTPServer阻塞的请求版本

import tornado.httpserver
import tornado.ioloop
import tornado.web
import base64
import socket

class MainHandler(tornado.web.RequestHandler):

def get(self):
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('localhost', 8888))
s.send('hello\n')
data = s.recv(4096)
self.write(data)
self.finish()

if __name__ == "__main__":
app = tornado.web.Application(
handlers = [
(r"/", MainHandler)
]
)

http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(8000)
tornado.ioloop.IOLoop.instance().start()

这里只是普通的socket请求，它是阻塞的。必须等2s后，服务器才会返回数据。

HTTPServer异步的请求版本

import tornado.httpserver
import tornado.web
import socket
from tornado import ioloop
from tornado import gen
from tornado.concurrent import Future

class MainHandler(tornado.web.RequestHandler):

@gen.coroutine
def get(self):
data = yield self.get_data()
self.write(data)
self.finish()

def get_data(self):
future = Future()
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('localhost', 8888))
s.send('hello\n')

def handle_data(sock, event):
io_loop = ioloop.IOLoop.current()
io_loop.remove_handler(sock)
data = sock.recv(1024)
future.set_result(data)

io_loop = ioloop.IOLoop.current()
io_loop.add_handler(s, handle_data, io_loop.READ)

return future

if __name__ == "__main__":
app = tornado.web.Application(
handlers = [
(r"/", MainHandler)
]
)

http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(8000)
tornado.ioloop.IOLoop.instance().start()

注意上面的异步函数get_data中，socket请求EchoServer，并不是直接recv等待结果。而是先实例化一个新的Future对象，

并且将socket可读事件登记到ioloop中。如果socket有数据时，就会回调handle_data，这里会调用future.set_result( )方法。

get_data满足了上面的两个条件:

返回Future对象

将future.set_result方法登记到ioloop中， 在结果返回时，就会调用。

最后注意gen.coroutine装饰器，如果被装饰的函数，出现异常，它是不会抛出的。需要访问返回的future，才能知道。

举例来说：

def handle_exce(future):
if future.exception() is not None:
print future.exec_info()

def func():
future = async()
future.add_done_callback(handle_exce)

@gen.coroutine
def async():
raise RuntimeError("async exception")

最后比较下性能，使用http_load测试：

这里模拟200个用户，不停的访问10s。

阻塞版

./http_load -p 200 -s 10  url
4 fetches, 200 max parallel, 20 bytes, in 10.0013 seconds
5 mean bytes/connection
0.399948 fetches/sec, 1.99974 bytes/sec
msecs/connect: 0.20175 mean, 0.277 max, 0.153 min
msecs/first-response: 5006.22 mean, 8010.14 max, 2002.34 min
HTTP response codes:
code 200 -- 4

异步版

./http_load -p 200 -s 10  url
800 fetches, 200 max parallel, 4000 bytes, in 10.0017 seconds
5 mean bytes/connection
79.9868 fetches/sec, 399.934 bytes/sec
msecs/connect: 20.0608 mean, 997.373 max, 0.03 min
msecs/first-response: 2020.97 mean, 2204.74 max, 2000.99 min
HTTP response codes:
code 200 -- 800

可以看到阻塞版的0.399948 fetches/sec和异步版的79.9853 fetches/sec。从中可见阻塞会大大影响tornado的性能。

比如耗时的数据库查询，耗时的运算。所以只有使用异步的库，才能发挥tornado的高性能。

如果没有相应的异步库，可以自己尝试着写。或者最简单的，使用celery作为异步任务队列。celery有对应的tornado库，tornado-celery。