当前位置：首页 > news >正文

Python 全栈系列220 Tornado的服务搭建

news 2025/9/15 17:47:15

说明

想法变的真快

本来是没打算用Tornado的，主要是想节约时间。但是现在看来不用还是不行：目前用gevent + flask部署的时候，启动多核的worker似乎存在问题。

另外，有很多内部基础的数据服务，其实并不需要flask的各种组件。

所以还是丢掉幻想，老老实实搭一个Tornado服务。

内容

1 背景

Tornado其实是我最早使用的Web服务框架，甚至现在还有一个服务是使用Tornado的，整体的效率真的很不错。

后来之所以用Flask，更多还是基于前端的考虑。

在确定了使用微服务体系后，其实应该直接再采用Tornado的，但当时有两个考虑：

1 Flask使用的还不熟，同时用两个脑子不够用
2 大部分情况下，Flask也是可以顶住的

现在看来，Web框架分为两类更为合理：一类面向流程，一般是低CPU耗用的应用，例如Portal，或者IO。另一类面向计算，一般会有更高的CPU耗用，而不用管流程上的问题（例如权限）

目前Flask那套，不说多么精通，至少很熟练了（不费精力就能轻易使用），所以现在倒不担心Tornado有啥额外负担。

另外，Tornado的定位很明确，就是进行内部的数据处理，不管任何权限类的事。所以可以构造非常简单的应用模型。

2 原理

先复习一下Tornado，以下内容参考这篇文章。

引用：
非阻塞式和基于Linux的Epoll（UNIX为kqueue）的异步网络IO异步非阻塞IO处理方式，单进程单线程异步IO的网络模型，可以编写异步非阻塞的程序非常适合开发长轮询、WebSocket和需要与每个用户建立持久连接的应用既是WebServer也是WebFramework

所以是适合处理处理的。

在这里插入图片描述

3 实现

先看看我的镜像里是否有包：

在这里插入图片描述

在老代码稍微改改就好了，突然回忆起了足够多的内容

'''
服务逻辑，字段设计
'''
# from server_funcs import *
import tornado.httpserver  # http服务器
import tornado.ioloop  # ?
import tornado.options  # 指定服务端口和路径解析
import tornado.web  # web模块
from tornado.options import define, options
import os.path  # 获取和生成template文件路径import json
from json import JSONEncoder
class MyEncoder(JSONEncoder):def default(self, obj):if isinstance(obj, np.integer):return int(obj)elif isinstance(obj, np.floating):return float(obj)elif isinstance(obj, np.ndarray):return obj.tolist()if isinstance(obj, datetime):return obj.__str__()if isinstance(obj, dd.timedelta):return obj.__str__()else:return super(MyEncoder, self).default(obj)# json.dumps(foo, cls=MyJsonEncoder)# 如果路径不存在则创建
def create_folder_if_notexist(somepath):if not os.path.exists(somepath):os.makedirs(somepath)return Truem_static = os.path.join(os.getcwd(),'m_static')
m_template = os.path.join(os.getcwd(),'m_template')create_folder_if_notexist(m_static)
create_folder_if_notexist(m_template)settings = {
'static_path':m_static,
'template_path':m_template
}
app_list = []IndexHandler_path = r'/'
class IndexHandler(tornado.web.RequestHandler):def get(self):self.write('【GET】This is Website for Internal API System')self.write('Please Refer to API document')print('Get got a request test')def post(self):request_body = self.request.bodyprint('Trying Decode Json')some_dict = json.loads(request_body)print(some_dict)msg_dict = {}msg_dict['info'] = '【POST】This is Website for Internal API System'msg_dict['input_dict'] = some_dictself.write(json.dumps(msg_dict))print('Post got a request test')
IndexHandler_tuple = (IndexHandler_path,IndexHandler)
app_list.append(IndexHandler_tuple)if __name__ == '__main__':tornado.options.parse_command_line()apps = tornado.web.Application(app_list, **settings)http_server = tornado.httpserver.HTTPServer(apps)define('port', default=8000, help='run on the given port', type=int)http_server.listen(options.port)tornado.ioloop.IOLoop.instance().start()