Python Asyncio run_in_executor创建的Future无法正确取消问题

asyncio.Future.cancel()的作用范围仅限事件循环本身，不会穿透到底层执行器：

• 调用cancel()后，事件循环只会把该Future标记为已取消状态，不再处理它的返回结果、不再执行绑定的回调函数，完全不会向ProcessPoolExecutor发送任何终止任务的指令。
• 提交到进程池的任务分两种状态：
  • 若任务还在进程池的等待队列中、未被worker进程拾取：部分Python版本会在Future被取消时自动把该任务从等待队列移除，这种场景下任务不会被执行。
  • 若任务已经被worker进程拾取、开始运行：任务的执行完全由子进程管控，和asyncio事件循环完全解耦，会一直运行到逻辑结束，哪怕关联的Future已经是cancelled状态，内存、CPU资源都会被持续占用。
    你遇到的就是第二种情况：任务已经开始在子进程里执行，仅取消asyncio层面的Future完全不会影响子进程的运行。

Python没有提供安全强制终止任意运行中函数的机制，要实现真正的任务取消，分两类方案：

方案1：协作式取消（生产环境推荐，无资源泄漏） #

这是最稳妥的实现方式：给提交到进程池的任务增加可感知的取消标记，任务执行过程中定期检查标记，收到取消信号后自行清理资源退出。
参考实现：

import asyncio
import functools
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
from typing import Callable

# 初始化进程池、跨进程共享状态管理器
executor = ProcessPoolExecutor(max_workers=5)
manager = Manager()
# 跨进程共享字典，存储每个任务的取消状态
cancel_flags = manager.dict()
task_id_counter = 0
list_of_futures = []

def wrap_task(func: Callable, task_id: int, *args, **kwargs):
    """包装用户提交的任务，注入取消检查逻辑"""
    try:
        # 给支持取消的函数传入cancel_check回调
        if "cancel_check" in func.__code__.co_varnames:
            kwargs["cancel_check"] = lambda: cancel_flags.get(task_id, False)
        return func(*args, **kwargs)
    finally:
        # 任务结束后清理对应标记
        if task_id in cancel_flags:
            del cancel_flags[task_id]

def run_in_another_process(func: Callable, *args, **kwargs) -> asyncio.Future:
    global task_id_counter
    loop = asyncio.get_running_loop()
    task_id = task_id_counter
    task_id_counter += 1
    # 初始化任务取消标记为未取消
    cancel_flags[task_id] = False
    future = loop.run_in_executor(
        executor,
        functools.partial(wrap_task, func, task_id, *args, **kwargs)
    )
    # 绑定task_id到future对象，方便取消时读取
    future.task_id = task_id
    list_of_futures.append(future)
    return future

async def cancel_task(fut: asyncio.Future):
    if fut.done():
        return
    # 第一步：标记asyncio层面的Future为已取消
    fut.cancel()
    # 第二步：设置跨进程取消标记，通知运行中的任务退出
    cancel_flags[fut.task_id] = True
    # 等待任务真正退出，避免遗留僵尸进程
    try:
        await fut
    except asyncio.CancelledError:
        pass

业务函数只需要在耗时步骤的间隙检查cancel_check()的返回值，返回True时直接终止逻辑、释放资源即可。

方案2：强制终止（仅适用于非核心场景，有资源泄漏风险） #

如果任务逻辑无法插入取消检查点（比如调用第三方阻塞库、无法修改源码），只能通过强制终止子进程的方式终止任务，但这种方案有明显缺陷：

• 被强制终止的子进程无法正常执行资源清理逻辑，可能导致文件句柄泄漏、锁未释放、临时文件残留、数据库连接断连等问题。
• 进程池的worker是复用的，直接杀掉worker进程会导致同一个worker上排队的其他无关任务也被终止。
  如果必须用这种方案，建议不要复用进程池worker，改为给每个任务单独启动独立子进程，记录子进程pid，取消时直接给对应pid发送终止信号，避免影响其他任务。

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