如何为flask_executor后台运行的线程或进程设置超时时间

flask_executor 本身没有内置你提到的两种超时配置能力，但可以通过简单的扩展实现对应效果，以下是可直接复用的实现方案：

1. 全局配置超时实现 #

我们可以自定义继承Executor的子类，读取全局配置的超时参数，自动绑定到所有提交的任务上：

from flask import Flask
from flask_executor import Executor
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from concurrent.futures import Future

# 先给Future类扩展set_timeout方法，兼容单任务超时配置场景
def set_timeout(self, timeout):
    self.timeout = timeout
Future.set_timeout = set_timeout

class TimeoutSupportExecutor(Executor):
    def _init_executor(self):
        # 读取全局超时配置，默认无超时
        self.global_timeout = self.app.config.get('EXECUTOR_TIMEOUT', None)
        if self.executor_type == 'thread':
            self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
        else:
            self.executor = ProcessPoolExecutor(max_workers=self.max_workers)
    
    def submit(self, fn, *args, **kwargs):
        future = super().submit(fn, *args, **kwargs)
        # 全局超时优先级低于单任务单独设置的超时
        if self.global_timeout and not hasattr(future, 'timeout'):
            future.set_timeout(self.global_timeout)
        return future

# 用法和你预期的完全一致
app = Flask(__name__)
app.config['EXECUTOR_TIMEOUT'] = 10 # 全局所有任务默认10秒超时
executor = TimeoutSupportExecutor(app)

2. 单任务单独设置超时实现 #

完成上述扩展后，即可直接使用你示例中的写法给单个任务设置独立超时：

def fib(n):
    if n <= 1:
        return n
    return fib(n-1) + fib(n-2)

@app.route('/run_fib')
def run_fib():
    future = executor.submit(fib, 5)
    future.set_timeout(10) # 该任务单独设置10秒超时，优先级高于全局配置
    return 'OK'

超时逻辑触发说明 #

如果需要主动清理超时任务，可以新增一个后台巡检线程实现：

import time
import threading

def cleanup_timeout_tasks():
    while True:
        for future in list(executor.futures):
            # 仅处理未完成且配置了超时的任务
            if hasattr(future, 'timeout') and not future.done():
                # 此处可以自己扩展记录任务提交时间的逻辑，判断是否超时
                is_timeout = True # 替换为实际的超时判断逻辑
                if is_timeout:
                    # 取消未运行的排队任务，已运行的任务仅进程模式可强制终止
                    future.cancel()
                    # 进程模式下可额外调用进程terminate方法强制杀死运行中超时进程
        time.sleep(1)

# 启动守护巡检线程
threading.Thread(target=cleanup_timeout_tasks, daemon=True).start()

注意事项 #

• 线程执行模式下Python无原生强制终止运行中线程的能力，超时配置仅能取消还未进入运行状态的排队任务，已启动的线程会继续运行，仅会忽略其返回结果
• 进程执行模式下可通过terminate方法强制终止超时进程，实现真正的运行中任务超时杀死
• 若仅需在获取任务返回结果时做超时控制，无需主动终止任务，可直接在调用future.result()时传入timeout参数：res = future.result(timeout=10)，超时后会抛出TimeoutError异常

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