阻塞和非阻塞Web框架
只有Tornado和Node.js是异步非阻塞的，其他所有的web框架都是阻塞式的。
Tornado阻塞和非阻塞两种模式都支持。 阻塞式：
代表：Django、Flask、Tornado、Bottle
一个请求到来未处理完成，后续请求则一直等待。
解决方案：多线程或多进程。 异步非阻塞（存在IO请求）：
代表：Tornado（默认单进程/单线程）
Tornado的阻塞模式示例
from tornado import ioloop
from tornado.web import RequestHandler,Application class IndexHandler(RequestHandler):
def get(self):
print("开始")
import time
time.sleep(10)
self.write("Hello, world")
print("结束") application = Application([
(r"/index", IndexHandler,{},'alias_name1'),
]) if __name__ == "__main__":
# 单进程
application.listen(80)
ioloop.IOLoop.instance().start() # 多进程
# from tornado.httpserver import HTTPServer
# server = HTTPServer(application)
# server.bind(8888)
# server.start(4) # Forks multiple sub-processes
# ioloop.IOLoop.current().start()
Tornado 异步非阻塞示例1
（注意：这个在window上跑有问题，还是阻塞的，和 time.time()有关？）
from tornado import ioloop
from tornado.web import RequestHandler,Application
from tornado import gen
from tornado.concurrent import Future
import time class IndexHandler(RequestHandler):
@gen.coroutine
def get(self):
print("开始")
future = Future()
ioloop.IOLoop.current().add_timeout(time.time() + 10, self.doing)
yield future def doing(self, *args, **kwargs):
self.write('async')
self.finish()
print("结束") application = Application([
(r"/index", IndexHandler),
]) if __name__ == "__main__":
application.listen(8000)
ioloop.IOLoop.instance().start()
Tornado 异步非阻塞示例2
from tornado import ioloop
from tornado.web import RequestHandler,Application
from tornado import gen
from tornado import httpclient class IndexHandler(RequestHandler):
@gen.coroutine
def get(self):
print("开始")
http = httpclient.AsyncHTTPClient()
yield http.fetch("https://github.com/", self.done) def done(self, response):
print(response)
self.write('执行成功')
self.finish()
print("结束") application = Application([
(r"/index", IndexHandler),
]) if __name__ == "__main__":
# 单进程
application.listen(9000)
ioloop.IOLoop.instance().start()
装饰器 + Future 实现Tornado的异步非阻塞示例一
from tornado import ioloop
from tornado.web import RequestHandler,Application
from tornado import gen
from tornado.concurrent import Future fu = None class IndexHandler(RequestHandler):
@gen.coroutine
def get(self):
print("I am coming...")
global fu
fu = Future()
fu.add_done_callback(self.done)
yield fu def done(self, response):
print(response.__dict__['_result'])
self.write("You made it.")
self.finish() class HelpHandler(RequestHandler):
@gen.coroutine
def get(self):
global fu
if isinstance(fu,Future):
fu.set_result("Just stick to it.")
self.write("Gave him a favor.")
else:
self.write("He already made it, don't worry.")
self.finish() application = Application([
(r"/index", IndexHandler),
(r"/help", HelpHandler),
]) if __name__ == "__main__":
application.listen(80)
ioloop.IOLoop.instance().start()
装饰器 + Future 实现Tornado的异步非阻塞示例二
from tornado import ioloop
from tornado.web import RequestHandler,Application
from tornado import gen
from tornado.concurrent import Future
from threading import Thread
import time def waiting(future):
time.sleep(5)
future.set_result('Hey man.') class IndexHandler(RequestHandler):
@gen.coroutine
def get(self):
fu = Future()
fu.add_done_callback(self.done) thread = Thread(target=waiting,args=(fu,))
thread.start() yield fu def done(self, response):
print(response.__dict__.get('_result'))
self.write("See you finally.")
self.finish() application = Application([
(r"/index", IndexHandler),
]) if __name__ == "__main__":
application.listen(80)
ioloop.IOLoop.instance().start()
基于异步非阻塞和Tornado-MySQL实现用户登录示例
"""
需要先安装支持异步操作Mysql的类库：
Tornado-MySQL： https://github.com/PyMySQL/Tornado-MySQL#installation
pip3 install Tornado-MySQL
"""
import tornado.web
from tornado import gen
import tornado_mysql
from tornado_mysql import pools POOL = pools.Pool(
dict(host='127.0.0.1', port=3306, user='admin', passwd='xxxxxx', db='school'),
max_idle_connections=1,
max_recycle_sec=3) @gen.coroutine
def get_user_by_conn_pool(user):
cur = yield POOL.execute("SELECT SLEEP(%s)", (user,))
row = cur.fetchone()
raise gen.Return(row) @gen.coroutine
def get_user(user):
conn = yield tornado_mysql.connect(host='8.8.8.8', port=3306, user='admin',
passwd='xxxxxx', db='school',charset='utf8')
cur = conn.cursor()
# yield cur.execute("SELECT name,age FROM student where name=%s", (user,))
yield cur.execute("select sleep(10)")
row = cur.fetchone()
cur.close()
conn.close()
raise gen.Return(row) class LoginHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.render('login.html') @gen.coroutine
def post(self, *args, **kwargs):
user = self.get_argument('name')
data = yield gen.Task(get_user, user)
if data:
print(data)
self.redirect('http://www.baidu.com')
else:
self.render('login.html') application = tornado.web.Application([
(r"/login", LoginHandler),
]) if __name__ == "__main__":
application.listen(8888)
tornado.ioloop.IOLoop.instance().start()
武Sir实现的异步非阻塞的微型Web框架 转载请注明出处
import re
import socket
import select
import time class HttpResponse(object):
"""
封装响应信息
"""
def __init__(self, content=''):
self.content = content self.headers = {}
self.cookies = {} def response(self):
return bytes(self.content, encoding='utf-8') class HttpNotFound(HttpResponse):
"""
404时的错误提示
"""
def __init__(self):
super(HttpNotFound, self).__init__('404 Not Found') class HttpRequest(object):
"""
用户封装用户请求信息
"""
def __init__(self, conn):
self.conn = conn self.header_bytes = bytes()
self.header_dict = {}
self.body_bytes = bytes() self.method = ""
self.url = ""
self.protocol = "" self.initialize()
self.initialize_headers() def initialize(self): header_flag = False
while True:
try:
received = self.conn.recv(8096)
except Exception as e:
received = None
if not received:
break
if header_flag:
self.body_bytes += received
continue
temp = received.split(b'\r\n\r\n', 1)
if len(temp) == 1:
self.header_bytes += temp
else:
h, b = temp
self.header_bytes += h
self.body_bytes += b
header_flag = True @property
def header_str(self):
return str(self.header_bytes, encoding='utf-8') def initialize_headers(self):
headers = self.header_str.split('\r\n')
first_line = headers[0].split(' ')
if len(first_line) == 3:
self.method, self.url, self.protocol = headers[0].split(' ')
for line in headers:
kv = line.split(':')
if len(kv) == 2:
k, v = kv
self.header_dict[k] = v class Future(object):
"""
异步非阻塞模式时封装回调函数以及是否准备就绪
"""
def __init__(self, callback):
self.callback = callback
self._ready = False
self.value = None def set_result(self, value=None):
self.value = value
self._ready = True @property
def ready(self):
return self._ready class TimeoutFuture(Future):
"""
异步非阻塞超时
"""
def __init__(self, timeout):
super(TimeoutFuture, self).__init__(callback=None)
self.timeout = timeout
self.start_time = time.time() @property
def ready(self):
current_time = time.time()
if current_time > self.start_time + self.timeout:
self._ready = True
return self._ready class Snow(object):
"""
微型Web框架类
"""
def __init__(self, routes):
self.routes = routes
self.inputs = set()
self.request = None
self.async_request_handler = {} def run(self, host='localhost', port=9999):
"""
事件循环
:param host:
:param port:
:return:
"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((host, port,))
sock.setblocking(False)
sock.listen(128)
sock.setblocking(0)
self.inputs.add(sock)
try:
while True:
readable_list, writeable_list, error_list = select.select(self.inputs, [], self.inputs,0.005)
for conn in readable_list:
if sock == conn:
client, address = conn.accept()
client.setblocking(False)
self.inputs.add(client)
else:
gen = self.process(conn)
if isinstance(gen, HttpResponse):
conn.sendall(gen.response())
self.inputs.remove(conn)
conn.close()
else:
yielded = next(gen)
self.async_request_handler[conn] = yielded
self.polling_callback() except Exception as e:
pass
finally:
sock.close() def polling_callback(self):
"""
遍历触发异步非阻塞的回调函数
:return:
"""
for conn in list(self.async_request_handler.keys()):
yielded = self.async_request_handler[conn]
if not yielded.ready:
continue
if yielded.callback:
ret = yielded.callback(self.request, yielded)
conn.sendall(ret.response())
self.inputs.remove(conn)
del self.async_request_handler[conn]
conn.close() def process(self, conn):
"""
处理路由系统以及执行函数
:param conn:
:return:
"""
self.request = HttpRequest(conn)
func = None
for route in self.routes:
if re.match(route[0], self.request.url):
func = route[1]
break
if not func:
return HttpNotFound()
else:
return func(self.request)
阻塞模式：基本使用
from snow import Snow
from snow import HttpResponse def index(request):
return HttpResponse('OK') routes = [
(r'/index/', index),
] app = Snow(routes)
app.run(port=8012)
异步非阻塞：超时
from snow import Snow
from snow import HttpResponse
from snow import TimeoutFuture request_list = [] def async(request):
obj = TimeoutFuture(5)
yield obj def home(request):
return HttpResponse('home') routes = [
(r'/home/', home),
(r'/async/', async),
] app = Snow(routes)
app.run(port=8012)
异步非阻塞：等待
from snow import Snow
from snow import HttpResponse
from snow import Future request_list = [] def callback(request, future):
return HttpResponse(future.value) def req(request):
obj = Future(callback=callback)
request_list.append(obj)
yield obj def stop(request):
obj = request_list[0]
del request_list[0]
obj.set_result('done')
return HttpResponse('stop') routes = [
(r'/req/', req),
(r'/stop/', stop),
] app = Snow(routes)
app.run(port=8012)