Azure Python函数应用：单实例超时终止不影响其他实例的实现方案

问题根源 #

你当前的实现使用os._exit()会直接终止整个Python进程，这会导致同一Azure实例上的所有函数执行任务都被终止——这就是两次执行都会被超时干掉的核心原因，因为它们共享同一个进程空间。另外Python的线程机制无法安全强制终止另一个线程（没有原生的terminate方法），所以用线程做超时监控根本无法实现只终止单个任务的需求。

解决方案：用子进程隔离单个任务 #

改用multiprocessing.Process来运行每个函数的execute方法，让每个任务都在独立的子进程中执行。超时后只需终止对应的子进程即可，不会影响主进程和其他子进程。

修改后的FAExecutor实现如下：

import multiprocessing
import time
from abc import ABC, abstractmethod

class FAExecutor(ABC):
    @abstractmethod
    def execute(self) -> None:
        raise NotImplementedError

    def execute_with_timeout(self) -> None:
        # 创建子进程运行当前任务的execute方法
        task_process = multiprocessing.Process(target=self.execute)
        task_process.start()
        
        # 等待任务执行，超时时间设为1200秒
        timeout = 1200
        task_process.join(timeout)
        
        # 如果超时后进程仍存活，强制终止并回收资源
        if task_process.is_alive():
            task_process.terminate()
            task_process.join()

class InfiteLoop(FAExecutor):
    def execute(self) -> None:
        logger = LoggerFactory.create_logger()

        while True:
            logger.warn("doing a looping")
            time.sleep(60)

关键说明 #

• multiprocessing.Process会创建完全独立的子进程，每个函数任务的运行环境相互隔离，终止子进程不会波及主进程或其他子进程。
• join(timeout)用于等待子进程执行完成，超时后通过is_alive()判断任务是否仍在运行，若超时则调用terminate()强制终止，最后再join()确保进程资源被彻底回收。
• 你的场景中死锁完全局限在单个函数内部，终止子进程不会遗留资源锁，这种方式是安全的，不会导致资源泄漏。

适配Azure Functions的注意事项 #

• 在Azure Functions Premium实例中，主进程负责管理多个函数执行的子进程，这种实现方式符合Azure的运行模型，不会干扰同一实例上的其他函数应用。
• 确保函数代码中没有跨进程共享的不可序列化资源（根据你的描述，死锁完全在单个函数内部，这一点无需额外调整）。

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