相较于asyncio无限循环，是否存在更优的KeyboardInterrupt捕获方式？

首先得明确：你当前用await loop.run_in_executor(None, thread.join)这类方法没法及时响应中断的核心原因——thread.join()是一个阻塞式调用，当它在executor里运行时，asyncio的事件循环会被卡住，没法处理KeyboardInterrupt这类信号，直到join完成（也就是线程结束）。要解决这个问题，我们得换思路，避免让事件循环被阻塞。

下面给你几个可行的方案，按推荐优先级排序：

1. 协作式线程终止（推荐，最安全） #

Python不支持强制终止线程（标准库没有thread.terminate()这类方法），因为强制终止会导致线程持有锁、打开的文件等资源无法释放，引发各种潜在问题。所以最佳实践是让长耗时进程主动检查终止信号，比如用threading.Event实现协作式停止。

修改你的代码如下：

import asyncio
import time
from threading import Thread, Event

def some_long_process(stop_event):
    # 把长任务拆成可中断的小步骤，定期检查停止信号
    for second in range(60):
        if stop_event.is_set():
            print("长进程收到终止信号，正在退出...")
            break
        time.sleep(1)
        print(f"进程已运行{second+1}秒")

async def main():
    stop_event = Event()
    thread = Thread(target=some_long_process, args=(stop_event,))
    thread.start()

    try:
        # 用asyncio.to_thread异步等待线程结束，不阻塞事件循环
        await asyncio.to_thread(lambda: thread.join())
    except KeyboardInterrupt:
        print("收到中断信号，通知线程停止...")
        stop_event.set()
        # 等待线程优雅退出
        thread.join()

if __name__ == '__main__':
    asyncio.run(main())

这个方案的好处是线程可以安全释放所有资源后退出，不会留下隐患。如果你的长任务是自己开发的，强烈建议用这种方式。

2. 用多进程替代线程（适合无法修改长任务的场景） #

如果你的长耗时进程是第三方库的函数，没法修改它来检查终止信号，那可以考虑用multiprocessing.Process代替threading.Thread。进程有独立的内存空间，支持安全的强制终止：

import asyncio
import time
from multiprocessing import Process

def some_long_process():
    time.sleep(60)

async def main():
    proc = Process(target=some_long_process)
    proc.start()

    try:
        await asyncio.to_thread(lambda: proc.join())
    except KeyboardInterrupt:
        print("收到中断信号，终止进程...")
        proc.terminate()
        # 等待进程完全退出
        proc.join()

if __name__ == '__main__':
    asyncio.run(main())

注意：进程间的通信和资源共享比线程麻烦，如果你的任务需要和主线程共享大量数据，这个方案的实现成本会高一些。

3. 封装轮询逻辑（最后备选） #

如果以上两种方案都不适用，那只能保留轮询，但可以把轮询逻辑封装成一个复用的异步函数，让代码更整洁：

import asyncio
import time
from threading import Thread

def some_long_process():
    time.sleep(60)

async def wait_for_thread(thread, check_interval=0.5):
    """异步等待线程结束，定期检查状态"""
    while thread.is_alive():
        await asyncio.sleep(check_interval)

async def main():
    thread = Thread(target=some_long_process)
    thread.start()

    try:
        await wait_for_thread(thread)
    except KeyboardInterrupt:
        print("收到中断信号，但无法安全终止线程，等待它自行结束...")
        thread.join()

if __name__ == '__main__':
    asyncio.run(main())

这个方案的局限性很明显：你没法强制终止线程，只能捕获中断后等待线程自己跑完，适合那些即使中断后也不影响系统状态的场景。

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