通用技术 python 的 web 框架 tornado 入门 (包含文档里忘记提到的那部分)
简介
Tornado 是一个用 Python 写的相对简单的、不设障碍的 Web 服务器架构，用以处理上万的同时的连接口，让实时的 Web 服务通畅起来。虽然跟现在的一些用 Python 写的 Web 架构相似，比如 Django，Flask，但 Tornado 更注重速度，能够处理海量的同时发生的流量。 最初由 FriendFeed 开发，后被 Facebook 收购，将其开源出来。
简单入门
pip install tornado
Hello world
这是一个非常简单的但是功能完善的程序
import tornado.ioloop
import tornado.web
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
def make_app():
return tornado.web.Application([
(r"/", MainHandler),
])
if __name__ == "__main__":
app = make_app()
app.listen(7000)
tornado.ioloop.IOLoop.current().start()
浏览器打开 http://localhost:7000就可以看到Hello, world了。
异步操作
异步操作需要 import gen 库 from tornado import gen
Sleep 操作
class MainHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
yield gen.sleep(5.0)
self.write("Hello, world")
请求外部链接
from tornado.httpclient import AsyncHTTPClient
class ProxyTesterhomeHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
body = yield self.get_testerhome()
self.write(body)
@gen.coroutine
def get_testerhome(self):
http_client = AsyncHTTPClient()
response = yield http_client.fetch("https://testerhome.com/")
raise gen.Return(response.body)
捕获 Ctrl+C 信号
运行那个 Hello world 样例的时候，你会发现无法用Ctrl+C结束掉那个应用，那是因为 tornado 默认不捕获 Interrupt 信号（文档里没有看到，Stackoverflow 上查到了）
try:
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
tornado.ioloop.IOLoop.instance().stop()
显示访问日志
下面这两行代码放到最上面
from tornado.log import enable_pretty_logging
enable_pretty_logging()
扩充make_app函数
设置默认模版的目录，静态文件目录，cookie 的 secret，登录的 url
def make_app(**settings):
settings['template_path'] = 'templates'
settings['static_path'] = 'static'
settings['cookie_secret'] = os.environ.get("SECRET", "SECRET:_")
settings['login_url'] = '/login'
return tornado.web.Application([
(r"/", MainHandler),
], **settings)
更多关于/login部分的内容，参考官网教程 http://tornado-zh.readthedocs.io/zh/latest/guide/security.html
渲染 html 文件
其中的settings['template_path'] = 'templates' 这个如果不设置，默认是当前目录。不过最好设置一下，不然显得文件放的很乱
目录结构
|-- index.html
`-- templates/
·-- index.html
index.html中的内容可以简单的写一下
<h1>Hello world</h1>
将原有的 MainHandler 改成
class MainHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
self.render("index.html")
访问首页的时候就可以看到刚才写的Hello world了
配置 X-Headers 支持
写完的程序一般会放到 Nginx 上，通过 Nginx 根据请求的域名转发给特定的后台服务。这时应用获取的请求 IP 就会是 Nginx 所在的 IP，想要获取真正的 IP 地址，就需要从 headers 里面取 X-Forwarded-For字段（这个字段是 nginx 设置的）。
简单的配置一下 tornado 程序，代码写起来就不会这么麻烦了。
import tornado.httpserver
app = make_app()
http_server = tornado.httpserver.HTTPServer(app, xheaders=True)
http_server.listen(7000)
# 替换掉 app.listen(7000)
开启 hot reload 模式
app = make_app(debug=True)
通过环境变量，来控制 hot reload 模式的开关
import os
hotreload = bool(os.getenv("DEBUG"))
app = make_app(debug=hotreload)
后台任务（threading)
参考: https://tornado-zh.readthedocs.io/zh/latest/guide/coroutines.html#id6
# coding: utf-8
import time
from concurrent.futures import ThreadPoolExecutor
# `pip install futures` for python2
from tornado.ioloop import IOLoop
from tornado.web import RequestHandler
from tornado.concurrent import run_on_executor
from tornado import gen
class Task():
_thread_pool = ThreadPoolExecutor(max_workers=4)
@run_on_executor(executor='_thread_pool') # default "executor"
def background_task(self, seconds):
print("Call started")
time.sleep(seconds)
return seconds
@gen.coroutine
def do(self, seconds=1):
val = yield self.background_task(seconds)
print("Return value:", val)
t = Task()
IOLoop.current().spawn_callback(t.do, .5)
IOLoop.current().run_sync(t.do)
期望输出
Call started
Call started
Return value: 0.5
Return value: 1
5.1 版本的 tornado 可以不用 gen.coroutine 改用 async 和 await 了
完整例子：复杂一点的 Hello world
# coding: utf-8
from __future__ import print_function
import os
import sys
import tornado.ioloop
import tornado.web
from tornado import gen
from tornado.httpclient import AsyncHTTPClient
from tornado.log import enable_pretty_logging
class MainHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
yield gen.sleep(.5)
self.write("Hello, world")
# save index.html into templates/
# self.render("index.html")
class ProxyTesterhomeHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
body = yield self.get_testerhome()
self.write(body)
@gen.coroutine
def get_testerhome(self):
http_client = AsyncHTTPClient()
response = yield http_client.fetch("https://testerhome.com/")
raise gen.Return(response.body)
def make_app(**settings):
settings['template_path'] = 'templates'
settings['static_path'] = 'static'
settings['cookie_secret'] = os.environ.get("SECRET", "SECRET:_")
settings['login_url'] = '/login'
return tornado.web.Application([
(r"/", MainHandler),
(r"/testerhome", ProxyTesterhomeHandler),
], **settings)
if __name__ == "__main__":
enable_pretty_logging()
port = 7000
app = make_app(debug=True)
print("Listening on *:{}".format(port))
http_server = tornado.httpserver.HTTPServer(app, xheaders=True)
http_server.listen(port)
# 3.8 changed the default event loop to ProactorEventLoop which doesn't
# implement everything required by tornado and breaks this benchmark.
# Restore the old WindowsSelectorEventLoop default for now.
# https://bugs.python.org/issue37373
# https://github.com/python/pyperformance/issues/61
# https://github.com/tornadoweb/tornado/pull/2686
if sys.platform == 'win32' and sys.version_info[:2] == (3, 8):
import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # 这里应该有一个引入时机问题，我记不太清了。
# Handle Ctrl-C on windows
if sys.platform == 'win32':
def _signal_handler(signum, frame):
print('Signal Interrupt catched, exiting...')
tornado.ioloop.IOLoop.instance().stop()
signal.signal(signal.SIGINT, _signal_handler)
tornado.ioloop.PeriodicCallback(lambda: None, 100).start()
try:
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
tornado.ioloop.IOLoop.instance().stop()
感谢
http://www.tornadoweb.org
https://github.com/codeskyblue/simple_tornado
如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！
