Pythonスレッディング：マルチスレッディング完全ガイドと実例

import threading
import time
 
def cpu_bound(n):
    """CPUバウンド: GILにより並列実行が妨げられる"""
    total = 0
    for i in range(n):
        total += i * i
    return total
 
def io_bound(url):
    """I/Oバウンド: ネットワーク待機中にGILが解放される"""
    import urllib.request
    return urllib.request.urlopen(url).read()
 
# CPUバウンド: 4つのスレッドが順番に実行される（スピードアップなし）
start = time.time()
threads = [threading.Thread(target=cpu_bound, args=(10_000_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"スレッドを使ったCPUバウンド処理: {time.time() - start:.2f}秒")
 
# I/Oバウンド: 4つのスレッドが待機時間を重ねて実行される（大幅なスピードアップ）

import threading
import time
 
def download_file(filename):
    print(f"[{threading.current_thread().name}] {filename}をダウンロード中...")
    time.sleep(2)  # ダウンロードをシミュレート
    print(f"[{threading.current_thread().name}] {filename}が完了")
 
# スレッドを作成
t1 = threading.Thread(target=download_file, args=("data.csv",))
t2 = threading.Thread(target=download_file, args=("report.pdf",))
 
# スレッドを開始
t1.start()
t2.start()
 
# 両方の終了を待つ
t1.join()
t2.join()
 
print("すべてのダウンロードが完了")

import threading
import time
 
def slow_task():
    time.sleep(10)
 
t = threading.Thread(target=slow_task)
t.start()
 
# 最大3秒待つ
t.join(timeout=3)
 
if t.is_alive():
    print("3秒後もスレッドは実行中")
else:
    print("スレッドが完了")

import threading
 
def worker():
    name = threading.current_thread().name
    print(f"実行中のスレッド: {name}")
 
t = threading.Thread(target=worker, name="DataProcessor")
t.start()
t.join()

import threading
import time
 
def background_monitor():
    while True:
        print("システム健全性を監視中...")
        time.sleep(5)
 
# デーモンスレッド: メインプログラム終了時に終了する
monitor = threading.Thread(target=background_monitor, daemon=True)
monitor.start()
 
# メインプログラムが作業を行う
time.sleep(12)
print("メインプログラムを終了")
# monitorスレッドは自動的に終了される

import threading
import time
 
class FileProcessor(threading.Thread):
    def __init__(self, filepath):
        super().__init__()
        self.filepath = filepath
        self.result = None
 
    def run(self):
        """run()をスレッドロジックでオーバーライド"""
        print(f"{self.filepath}を処理中")
        time.sleep(1)  # 作業をシミュレート
        self.result = f"処理済み: {self.filepath}"
 
# 作成して実行
processor = FileProcessor("/data/report.csv")
processor.start()
processor.join()
print(processor.result)

import threading
 
def fetch_data(url, timeout, retries=3, verbose=False):
    print(f"{url}を取得中 (タイムアウト={timeout}秒, 再試行={retries}, 詳細={verbose})")
 
# 位置引数をタプルとして
t1 = threading.Thread(target=fetch_data, args=("https://api.example.com", 30))
 
# キーワード引数を辞書として
t2 = threading.Thread(
    target=fetch_data,
    args=("https://api.example.com",),
    kwargs={"timeout": 30, "retries": 5, "verbose": True}
)
 
t1.start()
t2.start()
t1.join()
t2.join()

import threading
 
results = {}
lock = threading.Lock()
 
def compute(task_id, value):
    result = value ** 2
    with lock:
        results[task_id] = result
 
threads = []
for i in range(5):
    t = threading.Thread(target=compute, args=(i, i * 10))
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()
 
print(results)  # {0: 0, 1: 100, 2: 400, 3: 900, 4: 1600}

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
def fetch_url(url):
    time.sleep(1)  # ネットワークリクエストをシミュレート
    return f"{url}からのコンテンツ"
 
urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
 
with ThreadPoolExecutor(max_workers=3) as executor:
    # タスクを送信してFutureオブジェクトを取得
    future_to_url = {executor.submit(fetch_url, url): url for url in urls}
 
    # 完了順に結果を処理
    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
            print(f"{url}: {data}")
        except Exception as e:
            print(f"{url}で例外が発生: {e}")

from concurrent.futures import ThreadPoolExecutor
 
def process_item(item):
    return item.upper()
 
items = ["apple", "banana", "cherry", "date"]
 
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_item, items))
 
print(results)  # ['APPLE', 'BANANA', 'CHERRY', 'DATE']

from concurrent.futures import ThreadPoolExecutor, as_completed
 
def risky_task(n):
    if n == 3:
        raise ValueError(f"不正な入力: {n}")
    return n * 10
 
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(risky_task, i): i for i in range(5)}
 
    for future in as_completed(futures):
        task_id = futures[future]
        try:
            result = future.result(timeout=5)
            print(f"タスク {task_id}: {result}")
        except ValueError as e:
            print(f"タスク {task_id}が失敗: {e}")
        except TimeoutError:
            print(f"タスク {task_id}がタイムアウト")

from concurrent.futures import ThreadPoolExecutor
import time
 
def long_task(n):
    time.sleep(5)
    return n
 
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(long_task, i) for i in range(10)]
 
    # 保留中のタスクをキャンセル（実行中のタスクはキャンセルできない）
    for f in futures[4:]:
        cancelled = f.cancel()
        print(f"キャンセル済み: {cancelled}")

import threading
 
class BankAccount:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()
 
    def withdraw(self, amount):
        with self.lock:  # 一度に1つのスレッドのみ
            if self.balance >= amount:
                self.balance -= amount
                return True
            return False
 
    def deposit(self, amount):
        with self.lock:
            self.balance += amount
 
account = BankAccount(1000)
 
def make_transactions():
    for _ in range(100):
        account.deposit(10)
        account.withdraw(10)
 
threads = [threading.Thread(target=make_transactions) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
 
print(f"最終残高: {account.balance}")  # 常に1000

import threading
 
class SafeCache:
    def __init__(self):
        self._data = {}
        self._lock = threading.RLock()
 
    def get(self, key):
        with self._lock:
            return self._data.get(key)
 
    def set(self, key, value):
        with self._lock:
            self._data[key] = value
 
    def get_or_set(self, key, default):
        with self._lock:
            # これはget()を呼び出し、_lockも獲得する
            # RLockはこれを許可する; 通常のLockはデッドロックする
            existing = self.get(key)
            if existing is None:
                self.set(key, default)
                return default
            return existing

import threading
import time
 
# 最大3つの同時データベース接続を許可
db_semaphore = threading.Semaphore(3)
 
def query_database(query_id):
    with db_semaphore:
        print(f"クエリ {query_id}: 接続済み (アクティブ接続: {3 - db_semaphore._value})")
        time.sleep(2)  # クエリをシミュレート
        print(f"クエリ {query_id}: 完了")
 
threads = [threading.Thread(target=query_database, args=(i,)) for i in range(10)]
for t in threads: t.start()
for t in threads: t.join()

import threading
import time
 
data_ready = threading.Event()
shared_data = []
 
def producer():
    print("Producer: データを準備中...")
    time.sleep(3)
    shared_data.extend([1, 2, 3, 4, 5])
    print("Producer: データ準備完了、コンシューマーにシグナル送信")
    data_ready.set()
 
def consumer(name):
    print(f"Consumer {name}: データを待機中...")
    data_ready.wait()  # イベントがセットされるまでブロック
    print(f"Consumer {name}: データを取得 = {shared_data}")
 
threads = [
    threading.Thread(target=producer),
    threading.Thread(target=consumer, args=("A",)),
    threading.Thread(target=consumer, args=("B",)),
]
for t in threads: t.start()
for t in threads: t.join()

import threading
import time
import random
 
buffer = []
MAX_SIZE = 5
condition = threading.Condition()
 
def producer():
    for i in range(20):
        with condition:
            while len(buffer) >= MAX_SIZE:
                condition.wait()  # 空きができるまで待機
            item = random.randint(1, 100)
            buffer.append(item)
            print(f"生産: {item} (バッファサイズ: {len(buffer)})")
            condition.notify_all()
        time.sleep(0.1)
 
def consumer(name):
    for _ in range(10):
        with condition:
            while len(buffer) == 0:
                condition.wait()  # アイテムができるまで待機
            item = buffer.pop(0)
            print(f"Consumer {name}が消費: {item} (バッファサイズ: {len(buffer)})")
            condition.notify_all()
        time.sleep(0.15)
 
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer, args=("X",))
t3 = threading.Thread(target=consumer, args=("Y",))
t1.start(); t2.start(); t3.start()
t1.join(); t2.join(); t3.join()

import threading
import queue
import time
 
task_queue = queue.Queue()
results = queue.Queue()
 
def worker():
    while True:
        item = task_queue.get()  # アイテムが利用可能になるまでブロック
        if item is None:
            break
        result = item ** 2
        results.put(result)
        task_queue.task_done()
 
# 4つのワーカーを開始
workers = []
for _ in range(4):
    t = threading.Thread(target=worker, daemon=True)
    t.start()
    workers.append(t)
 
# タスクを送信
for i in range(20):
    task_queue.put(i)
 
# すべてのタスクが完了するのを待つ
task_queue.join()
 
# ワーカーを停止
for _ in range(4):
    task_queue.put(None)
for w in workers:
    w.join()
 
# 結果を収集
all_results = []
while not results.empty():
    all_results.append(results.get())
print(f"結果: {sorted(all_results)}")

from collections import deque
import threading
import time
 
buffer = deque(maxlen=1000)
 
def producer():
    for i in range(100):
        buffer.append(i)
        time.sleep(0.01)
 
def consumer():
    consumed = 0
    while consumed < 100:
        if buffer:
            item = buffer.popleft()
            consumed += 1
        else:
            time.sleep(0.01)
    print(f"{consumed}個のアイテムを消費")
 
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start(); t2.start()
t1.join(); t2.join()

import threading
import queue
import time
import random
 
def producer(q, name, num_items):
    for i in range(num_items):
        item = f"{name}-item-{i}"
        q.put(item)
        print(f"Producer {name}: {item}を作成")
        time.sleep(random.uniform(0.05, 0.15))
    print(f"Producer {name}: 完了")
 
def consumer(q, name, stop_event):
    while not stop_event.is_set() or not q.empty():
        try:
            item = q.get(timeout=0.5)
            print(f"Consumer {name}: {item}を処理中")
            time.sleep(random.uniform(0.1, 0.2))
            q.task_done()
        except queue.Empty:
            continue
    print(f"Consumer {name}: シャットダウン中")
 
task_queue = queue.Queue(maxsize=10)
stop_event = threading.Event()
 
producers = [
    threading.Thread(target=producer, args=(task_queue, "P1", 10)),
    threading.Thread(target=producer, args=(task_queue, "P2", 10)),
]
consumers = [
    threading.Thread(target=consumer, args=(task_queue, "C1", stop_event)),
    threading.Thread(target=consumer, args=(task_queue, "C2", stop_event)),
    threading.Thread(target=consumer, args=(task_queue, "C3", stop_event)),
]
 
for c in consumers: c.start()
for p in producers: p.start()
for p in producers: p.join()
 
task_queue.join()  # すべてのアイテムが処理されるのを待つ
stop_event.set()   # コンシューマーに停止をシグナル
for c in consumers: c.join()

import threading
import queue
 
class WorkerPool:
    def __init__(self, num_workers):
        self.task_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.workers = []
 
        for _ in range(num_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)
 
    def _worker(self):
        while True:
            func, args, kwargs, future_id = self.task_queue.get()
            if func is None:
                break
            try:
                result = func(*args, **kwargs)
                self.result_queue.put((future_id, result, None))
            except Exception as e:
                self.result_queue.put((future_id, None, e))
            finally:
                self.task_queue.task_done()
 
    def submit(self, func, *args, **kwargs):
        future_id = id(func)  # 単純なID
        self.task_queue.put((func, args, kwargs, future_id))
        return future_id
 
    def shutdown(self):
        for _ in self.workers:
            self.task_queue.put((None, None, None, None))
        for w in self.workers:
            w.join()
 
# 使用法
pool = WorkerPool(4)
for i in range(10):
    pool.submit(lambda x: x * x, i)
pool.task_queue.join()
pool.shutdown()

import threading
import time
from concurrent.futures import ThreadPoolExecutor
 
class RateLimiter:
    def __init__(self, max_per_second):
        self.interval = 1.0 / max_per_second
        self.lock = threading.Lock()
        self.last_call = 0
 
    def wait(self):
        with self.lock:
            elapsed = time.time() - self.last_call
            wait_time = self.interval - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            self.last_call = time.time()
 
limiter = RateLimiter(max_per_second=5)
 
def rate_limited_fetch(url):
    limiter.wait()
    print(f"{time.time():.2f}に{url}を取得中")
    time.sleep(0.5)  # リクエストをシミュレート
    return f"{url}からのデータ"
 
urls = [f"https://api.example.com/item/{i}" for i in range(20)]
 
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(rate_limited_fetch, urls))

import threading
 
# 悪い例: 競合状態
counter = 0
 
def increment_unsafe():
    global counter
    for _ in range(100_000):
        counter += 1  # 読み込み、インクリメント、書き込み: アトミックではない
 
threads = [threading.Thread(target=increment_unsafe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"期待値: 500000, 取得値: {counter}")  # 500000未満になることが多い
 
# 良い例: ロックで保護
counter = 0
lock = threading.Lock()
 
def increment_safe():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1
 
threads = [threading.Thread(target=increment_safe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"期待値: 500000, 取得値: {counter}")  # 常に500000

import threading
 
lock_a = threading.Lock()
lock_b = threading.Lock()
 
def thread_1():
    with lock_a:
        print("Thread 1: lock_aを獲得")
        with lock_b:  # thread_2がlock_bを保持している場合、永遠に待つ
            print("Thread 1: lock_bを獲得")
 
def thread_2():
    with lock_b:
        print("Thread 2: lock_bを獲得")
        with lock_a:  # thread_1がlock_aを保持している場合、永遠に待つ
            print("Thread 2: lock_aを獲得")
 
# これはデッドロックする
# t1 = threading.Thread(target=thread_1)
# t2 = threading.Thread(target=thread_2)
# t1.start(); t2.start()

def thread_1_fixed():
    with lock_a:    # 常にlock_aを最初に
        with lock_b:
            print("Thread 1: 両方のロックを獲得")
 
def thread_2_fixed():
    with lock_a:    # 常にlock_aを最初に（同じ順序）
        with lock_b:
            print("Thread 2: 両方のロックを獲得")

def safe_acquire():
    acquired_a = lock_a.acquire(timeout=2)
    if not acquired_a:
        print("lock_aを獲得できないため、バックオフ")
        return
    try:
        acquired_b = lock_b.acquire(timeout=2)
        if not acquired_b:
            print("lock_bを獲得できないため、lock_aを解放")
            return
        try:
            print("安全に両方のロックを獲得")
        finally:
            lock_b.release()
    finally:
        lock_a.release()

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import time
 
def fetch_page(url):
    """Webページを取得してコンテンツ長を返す"""
    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            content = response.read()
            return url, len(content), None
    except Exception as e:
        return url, 0, str(e)
 
urls = [
    "https://python.org",
    "https://docs.python.org",
    "https://pypi.org",
    "https://realpython.com",
    "https://github.com",
    "https://stackoverflow.com",
    "https://news.ycombinator.com",
    "https://httpbin.org",
]
 
# 順次実行
start = time.time()
for url in urls:
    fetch_page(url)
sequential_time = time.time() - start
 
# スレッドを使用した並列実行
start = time.time()
with ThreadPoolExecutor(max_workers=8) as executor:
    futures = {executor.submit(fetch_page, url): url for url in urls}
    for future in as_completed(futures):
        url, size, error = future.result()
        if error:
            print(f"  FAIL {url}: {error}")
        else:
            print(f"  OK   {url}: {size:,} bytes")
threaded_time = time.time() - start
 
print(f"\n順次実行: {sequential_time:.2f}秒")
print(f"スレッド使用:   {threaded_time:.2f}秒")
print(f"スピードアップ:    {sequential_time / threaded_time:.1f}倍")

from concurrent.futures import ThreadPoolExecutor
import os
import hashlib
 
def process_file(filepath):
    """ファイルを読み込んでSHA-256ハッシュを計算"""
    with open(filepath, 'rb') as f:
        content = f.read()
    file_hash = hashlib.sha256(content).hexdigest()
    size = os.path.getsize(filepath)
    return filepath, file_hash, size
 
def hash_all_files(directory, pattern="*.py"):
    """スレッドを使用してディレクトリ内の一致するすべてのファイルのハッシュを計算"""
    import glob
    files = glob.glob(os.path.join(directory, "**", pattern), recursive=True)
 
    results = {}
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = {executor.submit(process_file, f): f for f in files}
        for future in futures:
            try:
                path, hash_val, size = future.result()
                results[path] = {"hash": hash_val, "size": size}
            except Exception as e:
                print(f"{futures[future]}の処理中にエラー: {e}")
 
    return results
 
# 使用法
# file_hashes = hash_all_files("/path/to/project")

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request
import json
import time
 
def fetch_api(endpoint, max_retries=3, backoff=1.0):
    """指数バックオフ再試行でAPIエンドポイントを取得"""
    for attempt in range(max_retries):
        try:
            url = f"https://jsonplaceholder.typicode.com{endpoint}"
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=10) as response:
                data = json.loads(response.read())
                return {"endpoint": endpoint, "data": data, "error": None}
        except Exception as e:
            if attempt < max_retries - 1:
                wait = backoff * (2 ** attempt)
                time.sleep(wait)
            else:
                return {"endpoint": endpoint, "data": None, "error": str(e)}
 
endpoints = [f"/posts/{i}" for i in range(1, 21)]
 
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(fetch_api, ep) for ep in endpoints]
    results = [f.result() for f in futures]
 
elapsed = time.time() - start
success = sum(1 for r in results if r["error"] is None)
print(f"{elapsed:.2f}秒で{len(endpoints)}エンドポイント中{success}個を取得")

import threading
import time
 
class PeriodicTask:
    """バックグラウンドスレッドで固定間隔で関数を実行"""
    def __init__(self, interval, func, *args, **kwargs):
        self.interval = interval
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self._stop_event = threading.Event()
        self._thread = None
 
    def start(self):
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
 
    def _run(self):
        while not self._stop_event.is_set():
            self.func(*self.args, **self.kwargs)
            self._stop_event.wait(self.interval)
 
    def stop(self):
        self._stop_event.set()
        if self._thread:
            self._thread.join()
 
# 使用法
def check_health():
    print(f"{time.strftime('%H:%M:%S')}にヘルスチェック")
 
task = PeriodicTask(2.0, check_health)
task.start()
time.sleep(7)
task.stop()
print("停止")

import time
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
 
def io_task():
    time.sleep(0.2)
 
def cpu_task(n=2_000_000):
    return sum(i * i for i in range(n))
 
# スレッディング vs マルチプロセッシングのベンチマーク
NUM_TASKS = 20
 
# スレッディング - I/Oバウンド
start = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(lambda _: io_task(), range(NUM_TASKS)))
print(f"スレッディング (I/O): {time.time() - start:.2f}秒")
 
# スレッディング - CPUバウンド
start = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(lambda _: cpu_task(), range(NUM_TASKS)))
print(f"スレッディング (CPU): {time.time() - start:.2f}秒")