tornado_5_异步web服务

tornado_5_异步web服务

大部分Web应用（包括我们之前的例子）都是阻塞性质的，包括之前提到的案例，而tornado支持应用程序在等待第一个处理完成的过程中，让I/O循环打开以便服务于其他客户端，直到处理完成时启动一个请求并给予反馈，而不再是等待请求完成的过程中挂起进程

同步

例子1

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import datetime

from tornado.options import define, options
define("port", default=8004, help="run on the given port", type=int)

class IndexHandler(tornado.web.RequestHandler):
def get(self):
query = self.get_argument('q')
client = tornado.httpclient.HTTPClient()
response = client.fetch("http://www.baidu.com?wd={}".format(query))
body = response.body
result_count = len(body)
now = datetime.datetime.utcnow()
self.write("""
<div style="text-align: center">
<div style="font-size: 72px">%s</div>
<div style="font-size: 144px">%s</div>
<div style="font-size: 144px">%s</div>
<div style="font-size: 24px">tweets per second</div>
</div>""" % (query,result_count, now))

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()

主要使用到httpclient,让请求指定为同步请求

client = tornado.httpclient.HTTPClient()
response = client.fetch("http://www.baidu.com?wd={}".format(query))
body = response.body

测试： 在10秒内执行大约100个并发请求，命令如下

siege http://localhost:8000/?q=pants -c100 -t10s

同步缺陷：无论每个请求自身返回多么快，API往返都会以至于产生足够大的滞后，因为进程直到请求完成并且数据被处理前都一直处于强制挂起状态

异步

异步demo

例子

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import datetime
from tornado.options import define, options

define("port", default=8005, help="run on the given port", type=int)

class IndexHandler(tornado.web.RequestHandler):
@tornado.web.asynchronous # 在get方法的定义之前
def get(self):
query = self.get_argument('q')
client = tornado.httpclient.AsyncHTTPClient()
client.fetch("http://www.baidu.com?wd={}".format(query), callback=self.on_response)

def on_response(self, response):
body = response.body
result_count = len(body)
now = datetime.datetime.utcnow()
self.write("""
<div style="text-align: center">
<div style="font-size: 72px">%s</div>
<div style="font-size: 144px">%s</div>
<div style="font-size: 24px">%s</div>
<div style="font-size: 24px">tweets per second</div>
</div>""" % (self.get_argument('q'), result_count, now))
self.finish() # 回调方法结尾处调用

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()

测试： 在10秒内执行大约100个并发请求，命令如下

siege http://localhost:8000/?q=pants -c100 -t10s

结论，改成异步后，速度边快、单位时间内的响应变多

tornado.gen

异步生成器，当我们在处理多个请求的时候，如果都是异步，那么会在回调中回调，比如下面

def get(self):
client = AsyncHTTPClient()
client.fetch("http://example.com", callback=on_response)

def on_response(self, response):
client = AsyncHTTPClient()
client.fetch("http://another.example.com/", callback=on_response2)

def on_response2(self, response):
client = AsyncHTTPClient()
client.fe
4000
tch("http://still.another.example.com/", callback=on_response3)

def on_response3(self, response):
[etc., etc.]

因此引入了tornado.gen模块，可以提供一个更整洁的方式来执行异步请求,可以解决以上的问题

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import tornado.gen

import datetime

from tornado.options import define, options
define("port", default=8006, help="run on the given port", type=int)

class IndexHandler(tornado.web.RequestHandler):
@tornado.web.asynchronous
@tornado.gen.engine
def get(self):
query = self.get_argument('q')
client = tornado.httpclient.AsyncHTTPClient()
response = yield tornado.gen.Task(client.fetch, "http://www.baidu.com?wd={}".format(query))
body = response.body
result_count = len(body)
now = datetime.datetime.utcnow()
self.write("""
<div style="text-align: center">
<div style="font-size: 24px">%s</div>
<div style="font-size: 24px">%s</div>
<div style="font-size: 24px">%s</div>
<div style="font-size: 24px">tweets per second</div>
</div>""" % (query, result_count, now))
self.finish()

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()

gen的优点：在请求处理程序中返回HTTP响应，而不是回调函数中。因此，代码更易理解：所有请求相关的逻辑位于同一个位置。而HTTP请求依然是异步执行的，所以我们使用tornado.gen可以达到和使用回调函数的异步请求版本相同的性能

长轮询

所谓的”服务器推送”技术允许Web应用实时发布更新，同时保持合理的资源使用以及确保可预知的扩展。对于一个可行的服务器推送技术而言，它必须在现有的浏览器上表现良好。最流行的技术是让浏览器发起连接来模拟服务器推送更新。这种方式的HTTP连接被称为长轮询或Comet请求，长轮询意味着浏览器只需启动一个HTTP请求，其连接的服务器会有意保持开启。浏览器只需要等待更新可用时服务器”推送”响应。当服务器发送响应并关闭连接后，（或者浏览器端客户请求超时），客户端只需打开一个新的连接并等待下一个更新

例子：

# coding=utf-8
import tornado.web
import tornado.httpserver
import tornado.ioloop
import tornado.options
from uuid import uuid4

class ShoppingCart(object):
"""
ShoppingCart维护我们的库存中商品的数量
"""
totalInventory = 10
callbacks = []
carts = {}

def register(self, callback):
self.callbacks.append(callback)

def moveItemToCart(self, session):
if session in self.carts:
return

self.carts[session] = True
self.notifyCallbacks()

def removeItemFromCart(self, session):
if session not in self.carts:
return

del(self.carts[session])
self.notifyCallbacks()

def notifyCallbacks(self):
for c in self.callbacks:
self.callbackHelper(c)

self.callbacks = []

def callbackHelper(self, callback):
callback(self.getInventoryCount())

def getInventoryCount(self):
return self.totalInventory - len(self.carts)

class DetailHandler(tornado.web.RequestHandler):
"""
为每个页面请求产生一个唯一标识符，在每次请求时提供库存数量，并向浏览器渲染index.html模板
"""
def get(self):
session = uuid4()
count = self.application.shoppingCart.getInventoryCount()
self.render("index_poll.html", session=session, count=count)

class CartHandler(tornado.web.RequestHandler):
"""
CartHandler用于提供操作购物车的接口
"""
def post(self):
action = self.get_argument('action')
session = self.get_argument('session')

if not session:
self.set_status(400)
return

if action == 'add':
self.application.shoppingCart.moveItemToCart(session)
elif action == 'remove':
self.application.shoppingCart.removeItemFromCart(session)
else:
self.set_status(400)

class StatusHandler(tornado.web.RequestHandler):
"""
StatusHandler用于查询全局库存变化的通知
"""
@tornado.web.asynchronous  # 使得Tornado在get方法返回时不会关闭连接
def get(self):
# 我们使用self.async_callback包住回调函数以确保回调函数中引发的异常不会使RequestHandler关闭连接
self.application.shoppingCart.register(self.on_message)

def on_message(self, count):
self.write('{"inventoryCount":"%d"}' % count)
# 长轮询连接已经关闭，购物车控制器必须删除已注册的回调函数列表中的回调函数
self.finish()

class Application(tornado.web.Application):
def __init__(self):
self.shoppingCart = ShoppingCart()

handlers = [
(r'/', DetailHandler),
(r'/cart', CartHandler),
(r'/cart/status', StatusHandler)
]

settings = {
'template_path': 'templates',
'static_path': 'static'
}

tornado.web.Application.__init__(self, handlers, **settings)

if __name__ == '__main__':
tornado.options.parse_command_line()
app = Application()
server = tornado.httpserver.HTTPServer(app)
server.listen(8007)
tornado.ioloop.IOLoop.instance().start()

inventory.js

$(document).ready(function() {
document.session = $('#session').val();

setTimeout(requestInventory, 100);

$('#add-button').click(function(event) {
jQuery.ajax({
url: '//localhost:8007/cart',
type: 'POST',
data: {
session: document.session,
action: 'add'
},
dataType: 'json',
beforeSend: function(xhr, settings) {
$(event.target).attr('disabled', 'disabled');
},
success: function(data, status, xhr) {
$('#add-to-cart').hide();
$('#remove-from-cart').show();
$(event.target).removeAttr('disabled');
}
});
});

$('#remove-button').click(function(event) {
jQuery.ajax({
url: '//localhost:8007/cart',
type: 'POST',
data: {
session: document.session,
action: 'remove'
},
dataType: 'json',
beforeSend: function(xhr, settings) {
$(event.target).attr('disabled', 'disabled');
},
success: function(data, status, xhr) {
$('#remove-from-cart').hide();
$('#add-to-cart').show();
$(event.target).removeAttr('disabled');
}
});
});
});

function requestInventory() {
jQuery.getJSON('//localhost:8007/cart/status', {session: document.session},
function(data, status, xhr) {
$('#count').html(data['inventoryCount']);
//setTimeout(requestInventory, 0);
}
);
}

index_poll.html

<html>
<head>
<title>Burt's Books – Book Detail</title>
<script src="https://cdn.bootcss.com/jquery/3.2.1/jquery.min.js"></script>
<script src="{{ static_url('scripts/inventory.js') }}"
type="application/javascript"></script>
</head>

<body>
<div>

f1db
<h1>Burt's Books</h1>

<hr/>
<p><h2>The Definitive Guide to the Internet</h2>
<em>Anonymous</em></p>
</div>

<!--<img src="static/images/internet.jpg" alt="The Definitive Guide to the Internet" />-->

<hr />

<input type="hidden" id="session" value="{{ session }}" />
<div id="add-to-cart">
<p><span style="color: red;">Only <span id="count">{{ count }}</span>
left in stock! Order now!</span></p>
<p>$20.00 <input type="submit" value="Add to Cart" id="add-button" /></p>
</div>
<div id="remove-from-cart" style="display: none;">
<p><span style="color: green;">One copy is in your cart.</span></p>
<p><input type="submit" value="Remove from Cart" id="remove-button" /></p>
</div>
</body>
</html>

tornado.web.asynchronous 命中这个方法的 HTTP 请求就成为长连接，直到你调用 self.finish 发送响应之前，连接都在等待状态（self.render的实现最后调用的就是self.finish）

优点：长轮询在站点或特定用户状态的高度交互反馈通信中非常有用

缺点： 长轮询开发应用时，记住对于浏览器请求超时间隔无法控制是非常重要的。许多浏览器限制了对于打开的特定主机的并发请求数量，当有一个连接保持空闲时，剩下的用来下载网站内容的请求数量就会有限制。

WebSockets

websockets是一个新标准，目前只支持特定版本浏览器

例子：

#coding=utf-8
import tornado.web
import tornado.websocket
import tornado.httpserver
import tornado.ioloop
import tornado.options
from uuid import uuid4

class ShoppingCart(object):
totalInventory = 10
callbacks = []
carts = {}

def register(self, callback):
self.callbacks.append(callback)

def unregister(self, callback):
"""yichu"""
self.callbacks.remove(callback)

def moveItemToCart(self, session):
if session in self.carts:
return

self.carts[session] = True
self.notifyCallbacks()

def removeItemFromCart(self, session):
if session not in self.carts:
return

del (self.carts[session])
self.notifyCallbacks()

def notifyCallbacks(self):
"""
我们不需要在它们被通知后移除内部的回调函数列表。我们只需要迭代列表并调用带有当前库存量的回调函数
:return:
"""
for callback in self.callbacks:
callback(self.getInventoryCount())

def getInventoryCount(self):
return self.totalInventory - len(self.carts)

class DetailHandler(tornado.web.RequestHandler):
def get(self):
session = uuid4()
count = self.application.shoppingCart.getInventoryCount()
self.render("index_poll.html", session=session, count=count)

class CartHandler(tornado.web.RequestHandler):
def post(self):
action = self.get_argument('action')
session = self.get_argument('session')

if not session:
self.set_status(400)
return

if action == 'add':
self.application.shoppingCart.moveItemToCart(session)
elif action == 'remove':
self.application.shoppingCart.removeItemFromCart(session)
else:
self.set_status(400)

class StatusHandler(tornado.websocket.WebSocketHandler):
"""
open, close分别在连接打开和接收到消息时被调用

"""
def open(self):
self.application.shoppingCart.register(self.callback)

def on_close(self):
self.application.shoppingCart.unregister(self.callback)

def on_message(self, message):
"""
在实现中，我们在一个新连接打开时使用ShoppingCart类注册了callback方法，并在连接关闭时注销了这个回调函数。
我们依然使用了CartHandler类的HTTP API调用，因此不需要监听WebSocket连接中的新消息，所以on_message实现是空的
:param message:
:return:
"""
pass

def callback(self, count):
self.write_message('{"inventoryCount":"%d"}' % count)

class Application(tornado.web.Application):
def __init__(self):
self.shoppingCart = ShoppingCart()

handlers = [
(r'/', DetailHandler),
(r'/cart', CartHandler),
(r'/cart/status', StatusHandler)
]

settings = {
'template_path': 'templates',
'static_path': 'static'
}

tornado.web.Application.__init__(self, handlers, **settings)

if __name__ == '__main__':
tornado.options.parse_command_line()
app = Application()
server = tornado.httpserver.HTTPServer(app)
server.listen(8007)
tornado.ioloop.IOLoop.instance().start()

inventory.js修改方法requestInventory

function requestInventory() {
var host = 'ws://localhost:8007/cart/status';

var websocket = new WebSocket(host);

websocket.onopen = function (evt) { };
websocket.onmessage = function(evt) {
$('#count').html($.parseJSON(evt.data)['inventoryCount']);
};
websocket.onerror = function (evt) { };
}