Python でスレッドの例外を処理する

import threading

def buggy_task():
    raise ValueError("エラーが発生！")

thread = threading.Thread(target=buggy_task)
thread.start()
thread.join()

print("メインスレッドは正常に終了")
# ValueError の情報はコンソールに表示されるが、メインでは捕捉できない

import threading

def safe_task():
    try:
        raise ValueError("エラーが発生！")
    except Exception as e:
        print(f"スレッド内でキャッチ: {e}")

thread = threading.Thread(target=safe_task)
thread.start()
thread.join()

import threading

class TaskRunner:
    def __init__(self, func, *args, **kwargs):
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.result = None
        self.exception = None
    
    def run(self):
        try:
            self.result = self.func(*self.args, **self.kwargs)
        except Exception as e:
            self.exception = e

def divide(a, b):
    return a / b

runner = TaskRunner(divide, 10, 0)
thread = threading.Thread(target=runner.run)
thread.start()
thread.join()

if runner.exception:
    print(f"例外が発生: {runner.exception}")
else:
    print(f"結果: {runner.result}")

from concurrent.futures import ThreadPoolExecutor

def risky_task(n):
    if n == 0:
        raise ValueError("ゼロは許可されていません")
    return 100 / n

with ThreadPoolExecutor() as executor:
    futures = [executor.submit(risky_task, i) for i in range(3)]
    
    for future in futures:
        try:
            result = future.result()
            print(f"結果: {result}")
        except ValueError as e:
            print(f"エラー: {e}")

from concurrent.futures import ThreadPoolExecutor

def task():
    raise RuntimeError("問題発生")

with ThreadPoolExecutor() as executor:
    future = executor.submit(task)
    
    exception = future.exception()  # 例外を取得（再送出しない）
    if exception:
        print(f"例外が発生: {type(exception).__name__}: {exception}")

import threading
import sys

def thread_exception_handler(args):
    print(f"スレッド例外をキャッチ:")
    print(f"  スレッド: {args.thread.name}")
    print(f"  例外: {args.exc_type.__name__}: {args.exc_value}")

threading.excepthook = thread_exception_handler

def buggy_task():
    raise ValueError("テストエラー")

thread = threading.Thread(target=buggy_task, name="BuggyThread")
thread.start()
thread.join()

import threading
from queue import Queue

error_queue = Queue()

def worker(n):
    try:
        if n % 2 == 0:
            raise ValueError(f"偶数エラー: {n}")
        print(f"Worker {n}: 成功")
    except Exception as e:
        error_queue.put((n, e))

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# エラーを確認
while not error_queue.empty():
    n, error = error_queue.get()
    print(f"Worker {n} でエラー: {error}")