ThreadPoolExecutor线程中捕获异常的问题及解决方案咨询

问题描述 #

我编写了如下AsyncPillow类及test函数：

class AsyncPillow:
    def __init__(self, FUNC: Callable, ARGS: tuple, POOL: ThreadPoolExecutor = None):
        self.FUNC = FUNC
        self.ARGS = ARGS
        self.POOL = POOL if POOL else ThreadPoolExecutor(thread_name_prefix = 'thread', initializer = None, initargs = None)

    def Exceptions(self):
        print('callback was called!')
        if e := self._future.exception():
            print(e)

    async def Call(self):
        loop = asyncio.get_event_loop()
        try:
            self._future: asyncio.Future = loop.run_in_executor(self.POOL, self.FUNC, *self.ARGS)
            await self._future

            self._future.add_done_callback(self.Exceptions)
        except Exception as e: 
            print(e)


    @property
    def result(self):
        return self._result if self._result else None

def test():
 print(1/0)

调用线程时难以捕获错误：await self._future后代码不再继续，异常未被捕获，Exceptions回调也未执行。

修改为如下代码后：

try:
            self._future: asyncio.Future = loop.run_in_executor(self.POOL, self.FUNC, *self.ARGS)
            await self._future

            self._future.add_done_callback(self.Exceptions)
        except: 
            print('hello')

会输出hello，但我无法理解该行为——await self._future为什么会触发except分支？

请问如何修改代码，才能通过add_done_callback捕获异常，并在任务完成后获取future的结果？

问题解析 #

1. await触发except的原因：用run_in_executor创建的Future，会把线程中函数抛出的异常封装到自身内部。当你await这个Future时，Python会自动将封装的异常重新抛出，所以会进入except分支。
2. 回调未执行的原因：原代码里是在await之后才添加回调，若await因异常跳出try块，添加回调的代码根本不会执行；就算任务正常完成，await后Future已处于完成状态，此时添加的回调虽会立即执行，但原代码未给self._result赋值，result属性也拿不到有效结果。另外，原回调函数未接收Future参数，写法不符合add_done_callback的要求。

修改后的完整代码 #

from concurrent.futures import ThreadPoolExecutor
from typing import Callable
import asyncio

class AsyncPillow:
    def __init__(self, FUNC: Callable, ARGS: tuple, POOL: ThreadPoolExecutor = None):
        self.FUNC = FUNC
        self.ARGS = ARGS
        self.POOL = POOL if POOL else ThreadPoolExecutor(thread_name_prefix='thread')
        self._result = None  # 初始化结果存储变量

    def _handle_task_done(self, future: asyncio.Future):
        print('callback was called!')
        try:
            self._result = future.result()  # 获取任务结果，有异常时会抛出
        except Exception as e:
            print(f"捕获到异常: {e}")
            # 可根据需求在这里设置异常默认结果
            # self._result = None

    async def Call(self):
        loop = asyncio.get_event_loop()
        self._future: asyncio.Future = loop.run_in_executor(self.POOL, self.FUNC, *self.ARGS)
        # 先添加回调，再执行await，确保回调一定会被触发
        self._future.add_done_callback(self._handle_task_done)
        try:
            await self._future
        except Exception as e:
            # 此处可选择性捕获异常，回调已处理过异常逻辑
            print(f"Await阶段捕获异常: {e}")

    @property
    def result(self):
        return self._result

def test():
    print(1/0)

# 测试逻辑
async def main():
    ap = AsyncPillow(test, ())
    await ap.Call()
    print(f"最终结果: {ap.result}")

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

关键修改点 #

• 回调函数_handle_task_done添加future参数，直接通过参数获取完成的任务实例，避免依赖实例变量self._future
• 调整顺序：先调用add_done_callback再执行await，确保无论任务成功或失败，回调都会被触发
• 在回调中通过future.result()获取结果并捕获异常，同时给self._result赋值，保证result属性能拿到有效数据
• 初始化self._result，避免属性访问时出现未定义错误

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