【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 asyncio

class 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_executor
    def 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_executor
    def 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.websocket

class 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):
    # 子进程池的 Executor
    executor = PROCESS_POOL

    @run_on_executor  # 将该方法交给子进程处理
    def process_in_subprocess(self, data):
        """使用线程池处理任务"""
        result = ThreadedHandler().execute_task(data)  # 使用已创建的线程池
        return result

    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()

流程优化后的执行

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

测试

1. 启动服务器：

   python your_script.py
   

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

3. 返回示例：

   Processed by PID 12345: test