脚本被中断时如何立即停止ThreadPoolExecutor并丢弃所有待处理任务

原有代码挂起的核心原因 #

• KeyboardInterrupt的捕获逻辑位置错误：大部分时候按下Ctrl+C时，主线程正阻塞在as_completed的等待逻辑中，异常会直接抛出到最外层的try块，内层针对future.result()的异常捕获根本不会触发，导致你写的executor.shutdown(wait=False, cancel_futures=True)逻辑没有执行
• ThreadPoolExecutor默认创建的是非守护线程，主线程结束时会强制等待所有非守护线程执行完成才会退出，已经发起的阻塞IO请求（比如你代码里的requests.get）没法被Python线程机制主动终止，会一直等到请求超时或返回才会结束
• ThreadPoolExecutor的with上下文管理器默认会在退出时执行shutdown(wait=True)，就算你手动调用了shutdown，上下文的自动逻辑还是会等待所有线程结束

修改后的可运行代码 #

import requests
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

# 自定义线程工厂，将所有工作线程设为守护线程
def daemon_thread_factory():
    base_factory = threading.Thread
    def _worker(*args, **kwargs):
        thread = base_factory(*args, **kwargs)
        thread.daemon = True
        return thread
    return _worker

def get_parallel(arguments_list: list[dict]):
    executor = ThreadPoolExecutor(max_workers=25)
    executor._thread_factory = daemon_thread_factory()
    try:
        futures_buffer = [executor.submit(requests.get, **kwargs) for kwargs in arguments_list]
        for future in as_completed(futures_buffer):
            try:
                response = future.result()
                print(response.url)
                yield response.url, response.status_code, response.json()['args']
            except Exception as exception:
                yield exception
    except KeyboardInterrupt:
        yield 'KeyboardInterrupt received, stopping all tasks'
        # 立即关闭执行器，取消所有未执行的任务，不等待正在运行的任务
        executor.shutdown(wait=False, cancel_futures=True)
        return
    # 正常执行完成的情况下关闭执行器
    executor.shutdown(wait=True)

if __name__ == '__main__':
    arguments = [dict(url=f'https://httpbin.org/get?q={i}') for i in range(200)]
    for t in get_parallel(arguments):
        print(t)

关键修改点 #

• 自定义线程工厂，把所有工作线程设置为守护线程，主线程触发中断退出时，会直接终止所有还在运行的工作线程，不会等待阻塞的请求返回
• 将KeyboardInterrupt的捕获逻辑移到as_completed循环的外层，确保任何时候触发中断都能第一时间执行关闭逻辑
• 显式管理Executor的生命周期，不使用默认的with上下文管理器，避免触发默认的wait=True的shutdown逻辑
• 触发中断后立即调用shutdown(wait=False, cancel_futures=True)，取消所有还没进入运行状态的待处理任务

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