Tornado通过使用非阻塞网络1/0，可以扩展到数以万计的开放链接，非常适合 长时间轮询，WebSockets和其他需要与每个用户建立长期连接的应用程序。
特点
- 注重性能优越，速度快
- 解决高并发
- 异步非阻塞
- websockets 长连接
- 内嵌了HTTP服务器
- 单线程的异步网络程序，默认启动时根据CPU数量运行多个实例:利用CPU多核的优势
pip3 install tornado
文档
main.py
import tornado.web
import tornado.ioloop
from tornado.options import parse_command_line, options, define
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.write('hello tornado')
def create_app():
'''创建app, 注册路由与处理类'''
app = tornado.web.Application(handlers=[
(r'/', HelloHandler)
],
template_path='templates', # 模板文件的目录
static_path='static', # 告诉tornado 静态目录
static_url_prefix='/static/' # 浏览器访问时的url
)
return app
if __name__ == '__main__':
# define, options, parse_command_line 三个配合使用解析命令行
# 定义默认参数 这里定义默认启动端口，可通过命令行启动时修改port的值，改变端口
define('port', default=8080, type=int)
# 解析命令行
parse_command_line() # 获取命令行启动时 --xx 参数和值
# 创建app
app = create_app()
# 监听端口
app.listen(options.port)
# python main.py --port=8080 # options.port 的值为 8080
# 服务监听
tornado.ioloop.IOLoop.instance().start()
请求
get请求
# http://127.0.0.1:8080/?name=asd&name=qwe
class HelloHandler(tornado.web.RequestHandler):
def get(self):
# 获取请求头信息
user_agent = self.request.headers.get('User-Agent', 'Unknown')
name = self.request.arguments.get('name') # 当url有相同参数时，值为列表
# ['asd', 'qwe']
name = self.get_arguments('name') # 当url有相同参数时，值为列表
# ['asd', 'qwe']
name = self.get_argument('name')
name = self.get_query_argument('name')
name = self.get_query_arguments('name')
print(name) # qwe
self.write('hello word')
app = tornado.web.Application(handlers=[
(r'/', HelloHandler)
])# http://127.0.0.1:8080/api/user/12
class HelloHandler(tornado.web.RequestHandler):
def get(self, uid):
self.write(f'user id is: {uid}')
app = tornado.web.Application(handlers=[
(r'/api/user/(\d+)', HelloHandler)
])post请求
form-data请求的参数值在
- self.request.arguments
- self.request.body_arguments
json提交的数据再body中
- self.request.body # 需要json反序列化
urlencoded 请求的参数值在
- self.request.arguments
- self.request.body_arguments
class HelloHandler(tornado.web.RequestHandler):
def post(self):
# 获取form-data, urlencoded 参数
name = self.request.arguments.get('name')
name = self.get_argument('name')
name = self.get_arguments('name')
name = self.get_body_argument('name')
name = self.get_body_arguments('name')
# 获取json参数
print(json.loads(self.request.body))
self.write('post')响应
write返回字符串
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.write('hello tornado')write返回json
- 注意：自己手动序列化json方式 前端response headers 中的 Content_Type属性text/html,而采用write自动序列化方式，Content_Type属性为application/json
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
data = {
'name': 'alex',
'age': 20,
'hobby': ['python', 'go', 'java', 'c++']
}
# 自动序列化
self.write(data) # {"name": "alex", "age": 20, "hobby": ["python", "go", "java", "c++"]}
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
data = {
'name': 'alex',
'age': 20,
'hobby': ['python', 'go', 'java', 'c++']
}
# 手动json序列化， 需设置header: Content_Type属性text/html
self.write(json.dumps(data))set_header设置响应头
还有 clear_header，add_header 方法
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
data = {
'name': 'alex',
'age': 20,
'hobby': ['python', 'go', 'java', 'c++']
}
self.set_header("Content-Type","application/json;charset=UTF-8")
self.set_header("token", "xxxxxxxxxxxx")
self.write(data)set_default_headers
规范默认修改响应的头的位置
- 注意： 在http处理方法中再使用 self.set_headers() 方法 设置通用的name的值，会覆盖原先在set_default_headers() 中设置的值
class HelloHandler(tornado.web.RequestHandler):
def set_default_headers(self):
self.set_header("Content-Type","application/json;charset=UTF-8")
self.set_header("token", "xxxxxxxxxxxx")
def get(self, *args, **kwargs):
data = {
'name': 'alex',
'age': 20,
'hobby': ['python', 'go', 'java', 'c++']
}
self.set_header("token", "aaaaaaaaaaaa") # 替换set_default_headers中设置的 token
self.write(data)set_status设置响应状态码
参数：
- status--状态码的值为 int类型
- reason--对状态码的描述 str类型 如果reason值为None 则状态码的值必须为正常值
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.set_status(404)
self.write('error')redirect重定向
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.write('hello tornado')
class UserHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.redirect('/')
app = tornado.web.Application(handlers=[
(r'/', HelloHandler),
(r'/user', UserHandler)
])send_error
- 作用：可以跑出http错误状态码，默认为500，跑出错误后tornado会调用write_error()方法处理，并返回给浏览器界面
- 注意： 不执行send_error之后的内容
class HelloHandler(tornado.web.RequestHandler):
def get(self, *args, **kwargs):
self.send_error(500， reason="server error")
# send_error 之后的代码不会执行
self.write('hello tornado')
write_error(status_code,**kwargs)
- 用来处理send_error跑出来的额信息，并返回给浏览器错误页面
class HelloHandler(tornado.web.RequestHandler):
def write_error(self, status_code: int, **kwargs: Any):
'''一般是返回自定义页面'''
code = 200
if status_code == 404:
code = 404
self.write('资源错误')
elif status_code == 500:
code = 500
self.write('服务器错误')
self.set_status(code)
def get(self, *args, **kwargs):
if self.get_argument('name') != 'alex':
self.send_error(404)
self.write('hello tornado')finish
self.finish() 是 RequestHandler 的一个方法，用于明确地结束 HTTP 请求处理。当你调用 self.finish() 时，Tornado 会立即发送响应给客户端，并关闭与该请求相关的所有资源。这意味着在 self.finish() 被调用之后，你不能再向响应中写入任何数据。
self.finish() 的主要用途包括：
-
提前终止响应：如果你在处理过程中发现不需要进一步处理或已经完成了所有的处理逻辑，可以调用
self.finish()来立即发送响应并结束处理过程。 -
异步处理完成后的清理：在使用异步操作（如数据库查询、网络请求等）时，一旦操作完成，你可以调用
self.finish()来结束请求处理。 -
防止意外的数据写入：通过显式调用
self.finish()，你可以确保不会在无意中向已发送的响应中添加额外的数据，这有助于避免产生不完整或损坏的响应。
文件上传下载
class UploadHandler(tornado.web.RequestHandler):
def post(self):
# 检查是否上传了文件
if not self.request.files:
self.write("No file uploaded")
return
# 获取文件对象
file_info = self.request.files['file'][0]
filename, content_type = file_info['filename'], file_info['content_type']
body = file_info['body']
# 定义保存路径
upload_path = "files"
if not os.path.exists(upload_path):
os.makedirs(upload_path)
# 保存文件
with open(os.path.join(upload_path, filename), 'wb') as f:
f.write(body)
self.write(f"File {filename} has been uploaded successfully.")
class DownloadHandler(tornado.web.RequestHandler):
def get(self, filename):
# 文件路径
dir_path = "files/"
file_path = os.path.join(dir_path, filename)
# 检查文件是否存在
if not os.path.exists(file_path):
raise tornado.web.HTTPError(404, "File not found")
# 设置响应头
self.set_header('Content-Type', 'application/octet-stream')
self.set_header('Content-Disposition', f'attachment; filename="{filename}"')
# 读取并发送文件
with open(file_path, 'rb') as f:
while True:
data = f.read(16384) # 一次读取16KB
if not data:
break
self.write(data)
self.finish()
app = tornado.web.Application(handlers=[
(r'/upload', UploadHandler),
(r'/download/(.*)', DownloadHandler),
])返回图片
import os
import tornado.ioloop
import tornado.web
class MainHandler(tornado.web.RequestHandler):
def get(self):
# 显示包含图片链接的页面
self.render("index.html")
class ImageHandler(tornado.web.RequestHandler):
def get(self, filename):
# 图片路径
image_path = os.path.join("static", "images", filename)
# 检查图片是否存在
if not os.path.exists(image_path):
raise tornado.web.HTTPError(404, "Image not found")
# 设置响应头
self.set_header('Content-Type', 'image/jpeg') # 根据图片类型设置 Content-Type
with open(image_path, 'rb') as f:
while True:
data = f.read(16384) # 一次读取16KB
if not data:
break
self.write(data)
self.finish()
def make_app():
return tornado.web.Application([
(r"/", MainHandler),
(r"/image/(.*)", ImageHandler), # 匹配/image/后面的所有内容作为图片文件名
(r"/static/(.*)", tornado.web.StaticFileHandler, {"path": "static"}), # 静态文件服务
], template_path=os.path.join(os.path.dirname(__file__), "templates"))
if __name__ == "__main__":
app = make_app()
app.listen(8080)
print("Server started on port 8080")
tornado.ioloop.IOLoop.current().start()web框架-tornado操作数据库 - 简书
![[Linux]用户管理指令](https://img-blog.csdnimg.cn/img_convert/2477bfb7bb1a58399ce078152cc04023.png)
