Python 3.11.9中asyncio.to_thread处理异常时解释器冻结问题求助

问题原因 #

在Windows平台的Python 3.11.9中，asyncio.to_thread会创建原生线程执行阻塞函数（如time.sleep或socket.accept）。当按下Ctrl+C触发KeyboardInterrupt时：

• 信号仅会被主线程捕获，执行阻塞系统调用的子线程正处于内核态阻塞，Windows无法强制中断这类线程；
• 子线程无法退出，导致asyncio事件循环无法完成收尾工作，进而整个Python解释器冻结，主线程的异常处理逻辑也无法执行。

简洁修复方法 #

方法1：改造阻塞函数为可中断形式 #

针对阻塞操作，拆分为多次短阻塞并检查中断状态，避免线程长时间处于无法中断的内核态阻塞：

import asyncio
import time
import signal

# 全局中断标志
_interrupted = False

def interrupt_handler(signum, frame):
    global _interrupted
    _interrupted = True

async def func_1():
    while not _interrupted:
        print('print something..')
        await asyncio.sleep(0.5)

def block_func():
    # 将长sleep拆分为多次短sleep，每次检查中断标志
    for _ in range(1000):
        if _interrupted:
            break
        time.sleep(0.1)

async def func_2():
    print('start here')
    await asyncio.to_thread(block_func)
    print('end here')

async def main():
    signal.signal(signal.SIGINT, interrupt_handler)
    asyncio.create_task(func_1())
    await func_2()

try:
    asyncio.run(main())
except BaseException:
    global _interrupted
    _interrupted = True
    print('exception has occured')

方法2：用asyncio.wait_for包装并配合事件循环终止 #

通过asyncio.wait_for包装to_thread调用，同时在捕获异常时主动终止事件循环，确保任务能被正确取消：

import asyncio
import time

async def func_1():
    while True:
        try:
            print('print something..')
            await asyncio.sleep(0.5)
        except asyncio.CancelledError:
            break

async def func_2():
    print('start here')
    try:
        # 用wait_for包装，允许任务被取消
        await asyncio.wait_for(asyncio.to_thread(time.sleep, 100), timeout=None)
    except asyncio.CancelledError:
        print('task cancelled')
    print('end here')

async def main():
    task1 = asyncio.create_task(func_1())
    await func_2()
    task1.cancel()

try:
    loop = asyncio.get_event_loop_policy().get_event_loop()
    loop.run_until_complete(main())
except KeyboardInterrupt:
    # 主动取消所有任务并终止事件循环
    for task in asyncio.all_tasks(loop):
        task.cancel()
    loop.run_until_complete(asyncio.gather(*asyncio.all_tasks(loop), return_exceptions=True))
    loop.close()
    print('exception has occured')

针对socket.accept的特殊处理 #

如果实际场景是socket.accept，可以给socket设置超时时间，循环调用accept并检查中断：

import socket

def socket_accept_block(sock):
    sock.settimeout(0.1)  # 设置0.1秒超时
    while not _interrupted:
        try:
            return sock.accept()
        except socket.timeout:
            continue

内容的提问来源于stack exchange，提问作者Mark-2008