在使用 ThreadPoolExecutor 时，直接捕获 KeyboardInterrupt 往往失效，因为主线程被阻塞在 executor.map() 等待所有工作线程完成，而信号无法及时中断阻塞操作；需借助进程级隔离与主动轮询实现可靠中断。
在使用 `threadpoolexecutor` 时，直接捕获 `keyboardinterrupt` 往往失效，因为主线程被阻塞在 `executor.map()` 等待所有工作线程完成，而信号无法及时中断阻塞操作；需借助进程级隔离与主动轮询实现可靠中断。
在 Python 的多线程并发编程中，一个常见误区是认为 try...except KeyboardInterrupt: 能像单线程程序那样立即响应 Ctrl+C。然而，当使用 concurrent.futures.ThreadPoolExecutor 并调用阻塞式方法（如 executor.map()、executor.submit().result()）时，主线程会挂起等待所有任务完成，此时即使操作系统已将 SIGINT 信号送达，Python 解释器也无法在阻塞点立即触发异常——它会被延迟到所有线程任务真正结束之后才抛出，导致“按了 Ctrl+C 却无反应”的假象。
根本原因在于：KeyboardInterrupt 是主线程的信号，而 ThreadPoolExecutor 内部的 map() 方法底层依赖 queue.get() 等不可中断的阻塞调用，且工作线程中的无限循环（如 while True: pass）既不检查中断标志，也不响应信号（线程无法直接接收 SIGINT）。
✅ 正确解决方案是将整个 ThreadPoolExecutor 执行逻辑封装进独立子进程，由主进程负责监控并主动终止：
import time
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Process
def sleeper(n):
print(f"[Thread-{n}] Starting...")
# 模拟实际工作（避免空循环占用 CPU）
time.sleep(0.25)
print(f"[Thread-{n}] Done.")
return n
def run():
with ThreadPoolExecutor(max_workers=10) as executor:
# 注意：executor.map() 返回迭代器，需遍历才会真正执行
list(executor.map(sleeper, range(50))) # 强制消费全部结果
if __name__ == "__main__":
process = Process(target=run)
process.start()
try:
# 主进程持续轮询子进程状态
while process.is_alive():
time.sleep(0.1) # 轻量等待，避免忙循环
except KeyboardInterrupt:
print("\n⚠️ Received Ctrl+C — terminating worker process...")
process.terminate() # 强制终止子进程及其所有线程
process.join(timeout=2) # 最多等待 2 秒优雅退出
if process.is_alive():
process.kill() # 强制杀死（兜底）
print("❗ Force killed unresponsive process.")
print("✅ Shutdown complete.")
? 关键要点说明：
- 不要在 ThreadPoolExecutor 上层直接 try/except KeyboardInterrupt：它无法穿透底层阻塞调用；
- 避免无限空循环（如 while True: pass）：这不仅导致 Ctrl+C 失效，还会 100% 占用 CPU 核心，应改用 time.sleep() 或带条件退出的逻辑；
- Process.terminate() 是核心：它向子进程发送 SIGTERM，可立即中断其所有线程（包括阻塞中的 map()），比线程级控制更可靠；
- 务必调用 join() + kill() 兜底：防止僵尸进程残留，确保资源彻底释放。
? 进阶建议：若需更精细的线程级协作中断（如取消未完成任务），可结合 executor.shutdown(wait=False) 与 future.cancel()，但前提是任务本身支持中断检查（例如定期轮询 threading.Event.is_set()）。对于简单场景，进程级隔离仍是最健壮、最易理解的方案。
