如何将工作线程的错误向上传播至主线程

你的核心问题在于两点：一是工作线程的异常无法传递到主线程，二是主线程卡在迭代器的q.get()调用上无法响应异常。以下是两种可行的解决方案：

方案一：手动实现异常捕获与终止信号 #

通过线程安全的异常存储和终止事件，让主线程及时感知异常并停止，同时修改迭代器避免永久阻塞：

import threading
import queue

# 共享的异常存储和终止信号
error_queue = queue.Queue(maxsize=1)
stop_event = threading.Event()

q = queue.Queue()
s = set()
for i in range(5):
    q.put(i)
    s.add(i)


def yielder():
    while not stop_event.is_set():
        try:
            # 设置超时，避免一直阻塞在空队列上
            ret = q.get(timeout=0.1)
        except queue.Empty:
            continue
        if ret is None:
            return
        yield ret
    # 收到终止信号后直接退出
    return


def result(x):
    print(x)
    s.remove(x)
    if not s:
        q.put(None)


def do_work_fn(i):
    if i == 3:
        raise AssertionError("NO THREES!")
    return i


def worker_task(i, result_fn):
    try:
        result_fn(do_work_fn(i))
    except Exception as e:
        # 只保存第一个出现的异常
        if not error_queue.full():
            error_queue.put(e)
        # 触发全局终止信号
        stop_event.set()


def do_pooled_work(work, result_fn):
    threads = []
    try:
        for i in work:
            # 检查终止信号，有异常就停止创建新线程
            if stop_event.is_set():
                break
            t = threading.Thread(target=worker_task, args=[i, result_fn])
            t.start()
            threads.append(t)
        
        # 等待线程过程中定期检查异常
        for t in threads:
            t.join(timeout=0.1)
            if not error_queue.empty():
                stop_event.set()
                break
        
        # 抛出捕获到的异常
        if not error_queue.empty():
            raise error_queue.get()
    finally:
        # 清理操作：终止迭代器，清空任务队列
        stop_event.set()
        while not q.empty():
            try:
                q.get_nowait()
            except queue.Empty:
                pass
        q.put(None)


# 执行后会立即抛出异常，不会卡住
do_pooled_work(yielder(), result)

关键修改点： #

• 异常捕获：每个工作线程捕获异常后存入error_queue，同时触发stop_event通知主线程。
• 可中断迭代器：yielder()加入终止信号检查，并给q.get()设置超时，避免永久阻塞。
• 主线程异常检查：在创建线程和等待线程时，实时检测异常状态，一旦发现异常就停止后续操作并抛出。
• 资源清理：finally块中确保终止信号触发、队列清空，避免迭代器和线程残留。

方案二：使用concurrent.futures.ThreadPoolExecutor（更简洁） #

Python标准库的ThreadPoolExecutor自带异常处理机制，能更优雅地实现需求：

import concurrent.futures
import queue

q = queue.Queue()
s = set()
for i in range(5):
    q.put(i)
    s.add(i)


def yielder():
    while True:
        ret = q.get()
        if ret is None:
            return
        yield ret


def result(x):
    print(x)
    s.remove(x)
    if not s:
        q.put(None)


def do_work_fn(i):
    if i == 3:
        raise AssertionError("NO THREES!")
    return i


def do_pooled_work(work, result_fn):
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        try:
            for i in work:
                future = executor.submit(lambda x: result_fn(do_work_fn(x)), i)
                futures.append(future)
                # 实时检查已完成的任务是否有异常
                for done_future in concurrent.futures.as_completed(futures):
                    try:
                        done_future.result()
                    except Exception as e:
                        # 取消所有未完成的任务
                        for f in futures:
                            f.cancel()
                        # 清空队列，让迭代器退出
                        while not q.empty():
                            q.get_nowait()
                        q.put(None)
                        raise e
        finally:
            if not q.empty():
                q.put(None)


do_pooled_work(yielder(), result)

优势： #

• 无需手动管理线程生命周期，ThreadPoolExecutor自动处理线程创建和销毁。
• as_completed方法可以实时获取已完成的任务，一旦发现异常立即取消所有未完成任务，清理队列并抛出异常。

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