【Python3】Tornado6.4 高性能编程

https://pypi.tuna.tsinghua.edu.cn/simple/tornado/

在 Tornado 6.4 中，实现最高性能需要根据应用的特点进行选择。Tornado 作为一个异步非阻塞框架，本身适合处理高并发请求，但不同的编程方式有不同的性能侧重点。以下是几种常见的方式，并对它们的性能和适用场景进行了比较。

---

1. 异步 + 非阻塞 I/O（最高性能）

推荐方式：使用 async / await 进行全异步编程

Tornado 的核心优势在于 异步非阻塞 I/O。如果你的任务主要涉及 I/O 密集型操作（如网络请求、数据库查询等），采用 纯异步编程 可以达到最高性能。

示例代码：

import tornado.ioloop
import tornado.web
import asyncioclass MainHandler(tornado.web.RequestHandler):async def get(self):# 模拟一个异步 I/O 操作result = await self.async_task()self.write(result)async def async_task(self):# 异步等待 3 秒，模拟 I/O 操作（如网络请求）await asyncio.sleep(3)return "Task completed asynchronously"def make_app():return tornado.web.Application([(r"/", MainHandler)])if __name__ == "__main__":app = make_app()app.listen(8888)print("Server running on http://localhost:8888")tornado.ioloop.IOLoop.current().start()

优势：

• 非阻塞 I/O，不会因单个慢请求阻塞事件循环。
• 可以处理数千个并发连接，特别适合 高并发 I/O 操作。
• 内存占用较低。

适用场景：
适合 网络请求、数据库查询 等 I/O 密集型任务。

---

2. 异步 + 多进程（CPU 密集型任务）

对于 CPU 密集型任务（如大规模计算、图像处理），单纯的异步编程无法充分利用多核 CPU 的优势。此时，可以使用 ProcessPoolExecutor 将任务分发到多个子进程。

示例代码：

import tornado.ioloop
import tornado.web
from tornado.concurrent import run_on_executor
from concurrent.futures import ProcessPoolExecutor
import time# 子进程池，全局共享
PROCESS_POOL = ProcessPoolExecutor(max_workers=4)def cpu_intensive_task(data):time.sleep(5)  # 模拟耗时计算return f"Processed: {data}"class MainHandler(tornado.web.RequestHandler):executor = PROCESS_POOL  # 使用子进程池@run_on_executordef process_in_subprocess(self, data):return cpu_intensive_task(data)async def get(self):data = self.get_argument("data", "default")result = await self.process_in_subprocess(data)self.write(result)def make_app():return tornado.web.Application([(r"/", MainHandler)])if __name__ == "__main__":app = make_app()app.listen(8888)print("Server running on http://localhost:8888")tornado.ioloop.IOLoop.current().start()

优势：

• 使用多进程处理 CPU 密集型任务，提高计算性能。
• 每个子进程拥有独立的 GIL（全局解释器锁），不会因为 Python 的 GIL 限制而阻塞。

适用场景：
适合 数据处理、机器学习推理、图像处理 等 CPU 密集型任务。

---

3. 异步 + 多线程（适合轻量 CPU 密集型任务）

对于需要一些轻量的 CPU 密集型任务，同时又不想引入复杂的多进程机制，可以使用 ThreadPoolExecutor。

示例代码：

import tornado.ioloop
import tornado.web
from tornado.concurrent import run_on_executor
from concurrent.futures import ThreadPoolExecutor
import time# 线程池，全局共享
THREAD_POOL = ThreadPoolExecutor(max_workers=10)def lightweight_cpu_task(data):time.sleep(2)  # 模拟轻量 CPU 任务return f"Processed: {data}"class MainHandler(tornado.web.RequestHandler):executor = THREAD_POOL  # 使用线程池@run_on_executordef process_in_thread(self, data):return lightweight_cpu_task(data)async def get(self):data = self.get_argument("data", "default")result = await self.process_in_thread(data)self.write(result)def make_app():return tornado.web.Application([(r"/", MainHandler)])if __name__ == "__main__":app = make_app()app.listen(8888)print("Server running on http://localhost:8888")tornado.ioloop.IOLoop.current().start()

优势：

• 使用线程池处理轻量的 CPU 任务。
• 相对于多进程，线程池的创建和销毁开销较小。

适用场景：
适合 轻量 CPU 任务（如简单的计算）和不频繁的 I/O 操作。

---

4. WebSocket + Tornado（实时通讯）

如果需要 实时通信，如 聊天应用 或 数据推送服务，可以使用 Tornado 的 WebSocket 支持。

示例代码：

import tornado.ioloop
import tornado.web
import tornado.websocketclass WebSocketHandler(tornado.websocket.WebSocketHandler):def open(self):print("WebSocket opened")def on_message(self, message):self.write_message(f"You said: {message}")def on_close(self):print("WebSocket closed")def make_app():return tornado.web.Application([(r"/ws", WebSocketHandler)])if __name__ == "__main__":app = make_app()app.listen(8888)print("WebSocket server running on ws://localhost:8888/ws")tornado.ioloop.IOLoop.current().start()

优势：

• 低延迟的双向通信。
• 非常适合 实时数据推送 和 聊天系统。

---

哪种方式性能最高？

1. I/O 密集型任务：使用 异步 I/O (async / await) 最高效。
2. CPU 密集型任务：使用 多进程池 (ProcessPoolExecutor) 效率更高。
3. 轻量 CPU 任务：使用 线程池 (ThreadPoolExecutor) 合适。
4. 实时通信：使用 WebSocket 提供双向通信能力。

---

总结

• 最高性能：如果任务主要是 I/O 密集型，纯异步编程（async / await）是最优解，因其最大化了并发性能。
• 计算密集型：对于需要大量 CPU 的任务，采用 多进程池 是最佳选择。
• 轻量计算：若任务不频繁且较轻量，可以使用 线程池。

选择合适的方式取决于任务的性质：

• I/O 密集 → 纯异步 (async / await)
• CPU 密集 → 多进程 (ProcessPoolExecutor)
• 实时通信 → WebSocket

5.异步 + 进程池 + 线程池

代码

import tornado.ioloop
import tornado.web
from tornado.concurrent import run_on_executor
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time
import os# 子进程池，全局共享，最多允许 4 个进程
PROCESS_POOL = ProcessPoolExecutor(max_workers=4)# 每个子进程内的线程池，使用类变量确保线程池只初始化一次
class ThreadedHandler:thread_pool = ThreadPoolExecutor(max_workers=10)  # 线程池最多允许 10 个线程def execute_task(self, data):"""在线程池中执行耗时任务"""future = self.thread_pool.submit(long_running_task, data)return future.result()  # 阻塞直到任务完成# 模拟的耗时任务
def long_running_task(data):print(f"[PID {os.getpid()}] Processing: {data}")time.sleep(5)  # 模拟任务耗时return f"Processed by PID {os.getpid()}: {data}"class MainHandler(tornado.web.RequestHandler):# 子进程池的 Executorexecutor = PROCESS_POOL@run_on_executor  # 将该方法交给子进程处理def process_in_subprocess(self, data):"""使用线程池处理任务"""result = ThreadedHandler().execute_task(data)  # 使用已创建的线程池return resultasync def get(self):data = self.get_argument("data", "default")# 等待子进程中的任务完成result = await self.process_in_subprocess(data)self.write(result)def make_app():return tornado.web.Application([(r"/", MainHandler),])if __name__ == "__main__":app = make_app()app.listen(8888)print("Server running on http://localhost:8888")tornado.ioloop.IOLoop.current().start()

---

流程优化后的执行

1. Tornado 主进程：收到请求并分配给一个子进程处理。
2. 子进程：使用已经存在的 线程池 处理任务。
3. 任务完成：将结果返回给客户端。

---

测试

1. 启动服务器：

   python your_script.py
   

2. 访问：http://localhost:8888?data=test

3. 返回示例：

   Processed by PID 12345: test
   

---