Python线程池任务取消实战：如何正确使用cancel()中断执行中的函数

1次阅读

没有评论

共计 2571 个字符，预计需要花费 7 分钟才能阅读完成。

在实际开发中，我们经常会遇到需要中断线程池任务的场景。比如：

用户在前端点击了取消按钮
任务执行时间超过预期阈值
系统资源不足需要优先处理其他任务
业务流程变更导致某些任务不再需要执行

如果没有合理的取消机制，这些已经提交但不再需要的任务会继续消耗系统资源，甚至导致严重的内存泄漏问题。

Python 中常见的任务中断方式主要有三种：

标志位检查
在线程函数内部定期检查某个标志位
优点是实现简单，适用所有 Python 版本
缺点是无法立即中断阻塞操作
协程取消
使用 asyncio 的 cancel()方法
适合异步 IO 场景
缺点是不能直接用于线程池
ThreadPoolExecutor.cancel()
专门用于线程池任务的取消
只能取消尚未开始的任务
需要配合 Future 对象使用

只能取消 尚未开始执行 的任务
对正在运行的任务调用 cancel()会返回 False
已完成的任务无法取消

理解 cancel()需要先了解 Future 对象的生命周期：

PENDING -> RUNNING -> CANCELLED (取消成功)
PENDING -> RUNNING -> FINISHED (正常完成)
PENDING -> CANCELLED (任务被取消)

当 cancel()成功时，Future 会进入 CANCELLED 状态，并设置 cancelled()为 True。

下面是一个完整的可取消任务实现方案：

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Callable
import time

def interruptible_task(task_func: Callable, check_interval: float = 0.1) -> Callable:
    """
    包装器函数，使任务支持定期检查中断标志

    Args:
        task_func: 要执行的任务函数
        check_interval: 检查中断的时间间隔(秒)
    Returns:
        包装后的可中断函数
    """
    def wrapped(*args, **kwargs):
        while True:
            # 在这里实现你的实际任务逻辑
            result = task_func(*args, **kwargs)

            # 定期检查是否应该中断
            time.sleep(check_interval)
            # 可以通过某种方式检查外部中断信号

            # 这里只是示例，实际应根据业务需求实现
            if should_cancel():  # 假设的取消检查函数
                raise CancelledError("Task was cancelled")

            return result
    return wrapped

def run_with_timeout(executor: ThreadPoolExecutor, 
                    task: Callable, 
                    timeout: float,
                    *args, **kwargs) -> Any:
    """
    带超时限制的任务执行

    Args:
        executor: 线程池执行器
        task: 要执行的任务
        timeout: 超时时间(秒)
        *args: 任务参数
        **kwargs: 任务关键字参数
    Returns:
        任务结果
    Raises:
        TimeoutError: 任务超时时抛出
    """
    future = executor.submit(task, *args, **kwargs)
    try:
        return future.result(timeout=timeout)
    except TimeoutError:
        print(f"Task timed out after {timeout} seconds")
        # 尝试取消任务
        if future.cancel():
            print("Task cancelled successfully")
        else:
            print("Task could not be cancelled (already running)")
        raise

# 使用示例
if __name__ == "__main__":
    def long_running_task(seconds: int) -> str:
        print(f"Task running for {seconds} seconds")
        time.sleep(seconds)
        return f"Completed after {seconds} seconds"

    interruptible = interruptible_task(long_running_task)

    with ThreadPoolExecutor() as executor:
        try:
            result = run_with_timeout(executor, interruptible, 2.5, 5)
            print(result)
        except TimeoutError:
            print("Timeout handled gracefully")

即使任务被取消，也要确保资源被正确释放：

def task_with_resources():
    resource = acquire_resource()
    try:
        # 任务逻辑
        while not should_cancel():
            do_work(resource)
    finally:
        # 确保无论如何都会释放资源
        release_resource(resource)