如何将Tornado PeriodicCallback的计算任务调度到非主线程？

核心思路 #

Tornado的PeriodicCallback默认在IOLoop主线程执行，耗时计算会阻塞整个事件循环。解决办法是将calculator1和calculator2放到独立线程池执行，仅保留push任务在主线程（因为要与WebSocket交互，必须在IOLoop线程操作）。

具体实现步骤 #

1. 引入线程池 #

使用Python标准库的concurrent.futures.ThreadPoolExecutor创建线程池，负责执行耗时计算任务：

import tornado.web
import tornado.ioloop
from concurrent.futures import ThreadPoolExecutor

# 根据任务类型调整线程数：CPU密集型设为CPU核心数，IO密集型可适当增大
executor = ThreadPoolExecutor(max_workers=2)

2. 包装计算任务 #

编写包装函数，将计算任务提交到线程池，避免直接在IOLoop线程执行：

def calculator1():
    # 你的耗时计算逻辑
    pass

def calculator2():
    # 可能超过500ms的耗时计算逻辑
    pass

def push():
    # 与WebSocket交互的推送逻辑（必须在IOLoop线程执行）
    pass

# 包装calculator1，提交到线程池
def run_calculator1():
    executor.submit(calculator1)

# 包装calculator2，提交到线程池
def run_calculator2():
    executor.submit(calculator2)

3. 调整PeriodicCallback的回调 #

将原来的计算任务替换为包装后的函数，确保IOLoop线程仅做调度，不执行耗时计算：

def make_app():
    return tornado.web.Application([
        (r"/websocket", WebsocketHandler),
        (r"/getData", DataRequestHandler),
    ])

if __name__ == '__main__':
    tornado.ioloop.PeriodicCallback(callback=run_calculator1, callback_time=500).start()
    tornado.ioloop.PeriodicCallback(callback=run_calculator2, callback_time=1000).start()
    tornado.ioloop.PeriodicCallback(callback=push, callback_time=1000).start()
    
    # 注册优雅关闭逻辑（可选）
    def on_shutdown():
        executor.shutdown(wait=True)
        tornado.ioloop.IOLoop.current().stop()
    
    tornado.ioloop.IOLoop.current().add_callback_from_signal(on_shutdown)
    
    tornado.ioloop.IOLoop.current().start()

处理计算结果（可选） #

如果计算任务的结果需要传递给主线程（比如供push任务使用），必须将结果处理逻辑切回IOLoop线程，避免线程安全问题：

def run_calculator_with_result():
    # 提交任务并获取Future对象
    future = executor.submit(calculator1)
    # 任务完成后，在IOLoop线程处理结果
    future.add_done_callback(lambda f: tornado.ioloop.IOLoop.current().add_callback(process_result, f.result()))

def process_result(result):
    # 处理计算结果，比如更新全局数据（此函数在IOLoop线程执行）
    global shared_data
    shared_data = result

关键注意事项 #

• 线程池大小：CPU密集型任务建议线程数等于CPU核心数；IO密集型任务可根据需求增大。
• 线程安全：共享数据的读写需加锁（如threading.Lock），或统一在IOLoop线程处理数据更新。
• WebSocket操作：所有与WebSocket相关的逻辑必须在IOLoop线程执行，严禁在子线程直接操作WebSocket连接。

内容的提问来源于stack exchange，提问作者BattosaiX