如何优雅关闭Tornado IOLoop？解决Ctrl+C触发的TimeoutError问题

我用Tornado的IOLoop实现了异步Worker功能，尝试通过Ctrl+C优雅关闭事件循环时，遇到了如下错误：

tornado.ioloop.TimeoutError: Operation timed out after None seconds

我知道可以捕获该错误，但希望以优雅的方式完成进程收尾，请问该如何实现？

相关代码片段：

#!/usr/bin/env python
import time
import signal
import random
from tornado import gen, ioloop, queues
concurrency = 10
def sig_exit(signum, frame):
    ioloop.IOLoop.current().add_callb...

嘿，这个问题我熟！你遇到的TimeoutError本质是关闭流程没妥善处理现有异步任务——要么没等任务做完就急着停IOLoop，要么关闭逻辑的线程安全没做好。下面给你一套完整的优雅关闭方案，直接适配你的代码：

1. 核心思路：标记关闭→停止新任务→等待旧任务→优雅收尾 #

我们需要按这几步来实现：

• 设置全局标记，通知Worker不再接受新任务
• 在信号处理函数里触发安全的关闭流程（必须在IOLoop线程执行，避免线程安全问题）
• 等待队列中现有任务完成，超时后强制取消剩余任务
• 捕获任务取消信号，做好资源释放等收尾

2. 完整修改后的代码 #

#!/usr/bin/env python
import time
import signal
import random
from tornado import gen, ioloop, queues

concurrency = 10
is_shutting_down = False
q = queues.Queue(maxsize=concurrency)

def sig_exit(signum, frame):
    global is_shutting_down
    is_shutting_down = True
    # 停止队列接受新任务
    q.stop()
    # 用add_callback_from_signal确保关闭操作在IOLoop线程执行（线程安全）
    ioloop.IOLoop.current().add_callback_from_signal(shutdown)

@gen.coroutine
def shutdown():
    try:
        # 等待所有任务完成，设置超时时间（可根据你的任务耗时调整，比如10秒）
        yield q.join(timeout=10)
    except ioloop.TimeoutError:
        # 超时后主动取消剩余未完成的任务
        for task in q._tasks:
            task.cancel()
        # 给点时间让取消操作完成
        yield gen.sleep(1)
    finally:
        # 安全关闭IOLoop
        ioloop.IOLoop.current().stop()

@gen.coroutine
def worker():
    # 只要没完全关闭，或者队列还有任务，就继续处理
    while not is_shutting_down or not q.empty():
        try:
            item = yield q.get()
            # 这里替换成你的实际任务处理逻辑
            yield gen.sleep(random.uniform(0.1, 1))
            print(f"Processed item: {item}")
        except gen.CancelledError:
            # 处理任务被取消的情况，比如释放资源、记录日志
            print("Task cancelled during graceful shutdown")
        finally:
            # 标记任务完成，不管成功还是取消
            q.task_done()

@gen.coroutine
def main():
    # 启动指定数量的Worker
    for _ in range(concurrency):
        gen.coroutine(worker)()
    # 模拟添加任务（实际场景可替换成你的任务来源）
    for i in range(100):
        if not is_shutting_down:
            yield q.put(i)
            time.sleep(0.01)

if __name__ == "__main__":
    # 注册信号处理：Ctrl+C触发SIGINT，kill命令触发SIGTERM
    signal.signal(signal.SIGINT, sig_exit)
    signal.signal(signal.SIGTERM, sig_exit)
    ioloop.IOLoop.current().run_sync(main)

3. 关键细节解释 #

• 线程安全的信号处理：用add_callback_from_signal而非直接在信号处理函数里操作IOLoop，因为信号处理是在单独线程执行的，直接操作会引发线程安全问题。
• 队列停止与任务等待：q.stop()会让后续的q.put抛出异常，阻止新任务进入；q.join(timeout)等待所有已入队的任务完成，超时后主动取消剩余任务，避免进程无限等待。
• 任务取消处理：Worker里捕获CancelledError，可以在这里做资源释放、日志记录等收尾工作，避免报错。
• 超时控制：设置timeout是为了平衡优雅性和可靠性——既给足时间让正常任务完成，又能防止卡死任务拖垮进程。

这样修改后，你按Ctrl+C时，进程会先停止接受新任务，等待现有任务完成，超时后自动取消剩余任务，最后安全关闭IOLoop，不会再出现恼人的TimeoutError啦！

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