import threading
<p>def faulty_task():
raise ValueError("出错了！")</p><p>t = threading.Thread(target=faulty_task)
t.start()
t.join()  # 即使join，也不会抛出异常
print("程序继续运行")  # 这行仍会执行</p>

import threading
import queue
<p>def worker(result_queue):
try:</p><h1>模拟任务</h1><pre class='brush:python;toolbar:false;'>    1 / 0
except Exception as e:
    result_queue.put(e)
else:
    result_queue.put(None)  # 成功完成

result_q = queue.Queue() t = threading.Thread(target=worker, args=(result_q,)) t.start() t.join()

主线程检查结果

exc = result_q.get() if exc: raise exc # 在主线程重新抛出

def safe_run(func, *args, **kwargs):
    try:
        return func(*args, **kwargs), None
    except Exception as e:
        return None, e
<p>def my_task(x, y):
return x / y</p><h1>使用示例</h1><p>result_q = queue.Queue()</p><p>def thread_wrapper():
result_q.put(safe_run(my_task, 5, 0))</p><p>t = threading.Thread(target=thread_wrapper)
t.start()
t.join()</p><p>result, exc = result_q.get()
if exc:
print(f"任务失败：{exc}")
else:
print(f"结果：{result}")</p>

from concurrent.futures import ThreadPoolExecutor
import time
<p>def bad_task():
time.sleep(1)
raise RuntimeError("任务崩溃")</p><p>with ThreadPoolExecutor() as executor:
future = executor.submit(bad_task)
try:
future.result()  # 此处会重新抛出异常
except RuntimeError as e:
print(f"捕获异常：{e}")</p>