Python concurrent.futures 實戰：Thread、Process 與 Interpreter Pool

python --version
python -c "import concurrent.futures; print('ready')"

from concurrent.futures import ThreadPoolExecutor
import time
def normalize_name(name: str) -> str:
    time.sleep(0.05)
    return name.strip().title()
with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(normalize_name, "  拍拍醬  ")
    print(future.running() or future.done())
    print(future.result(timeout=1))

from concurrent.futures import ThreadPoolExecutor
import time
def delayed_square(number: int) -> int:
    time.sleep((4 - number) * 0.02)
    return number * number
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(delayed_square, [1, 2, 3]))
print(results)  # [1, 4, 9]，維持輸入順序

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def fetch(item: str, delay: float) -> str:
    time.sleep(delay)
    return f"{item}:ok"
jobs = {"alpha": 0.08, "beta": 0.02, "gamma": 0.05}
with ThreadPoolExecutor(max_workers=3) as executor:
    future_to_name = {
        executor.submit(fetch, name, delay): name
        for name, delay in jobs.items()
    }
    for future in as_completed(future_to_name):
        name = future_to_name[future]
        print(name, future.result())

from concurrent.futures import ProcessPoolExecutor
import math
def is_prime(number: int) -> bool:
    if number < 2:
        return False
    if number == 2:
        return True
    if number % 2 == 0:
        return False
    limit = math.isqrt(number)
    return all(number % divisor for divisor in range(3, limit + 1, 2))
def main() -> None:
    numbers = [999_983, 1_000_003, 1_000_033]
    with ProcessPoolExecutor() as executor:
        results = executor.map(is_prime, numbers, chunksize=1)
        print(list(zip(numbers, results)))
if __name__ == "__main__":
    main()

import concurrent.futures as futures
import math
def count_factors(number: int) -> int:
    limit = math.isqrt(number)
    count = 0
    for candidate in range(1, limit + 1):
        if number % candidate == 0:
            count += 1 if candidate * candidate == number else 2
    return count
def main() -> None:
    executor_type = getattr(futures, "InterpreterPoolExecutor", None)
    if executor_type is None:
        raise RuntimeError("這個範例需要 Python 3.14+")
    numbers = [10_000_019, 10_000_079, 10_000_103]
    with executor_type(max_workers=3) as executor:
        print(list(executor.map(count_factors, numbers)))
if __name__ == "__main__":
    main()

from concurrent.futures import ThreadPoolExecutor, as_completed
def parse_score(raw: str) -> int:
    score = int(raw)
    if not 0 <= score <= 100:
        raise ValueError(f"score out of range: {score}")
    return score
records = ["80", "oops", "105", "92"]
successes: list[tuple[str, int]] = []
failures: list[tuple[str, str]] = []
with ThreadPoolExecutor(max_workers=3) as executor:
    future_to_raw = {
        executor.submit(parse_score, raw): raw for raw in records
    }
    for future in as_completed(future_to_raw):
        raw = future_to_raw[future]
        try:
            successes.append((raw, future.result()))
        except (TypeError, ValueError) as exc:
            failures.append((raw, str(exc)))
print("ok", sorted(successes))
print("failed", sorted(failures))

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time
def slow_job() -> str:
    time.sleep(0.2)
    return "finished"
with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(slow_job)
    try:
        print(future.result(timeout=0.05))
    except TimeoutError:
        print("等待逾時，但工作可能仍在背景執行")
    print(future.result(timeout=1))

from concurrent.futures import FIRST_EXCEPTION, ThreadPoolExecutor, wait
import time
def transform(value: int) -> int:
    time.sleep(0.01)
    if value == 3:
        raise ValueError("bad input: 3")
    return value * 10
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(transform, value) for value in range(8)]
    done, not_done = wait(futures, return_when=FIRST_EXCEPTION)
    failed = [future for future in done if future.exception() is not None]
    if failed:
        for future in not_done:
            future.cancel()  # 只能取消尚未開始者
        print("first error:", failed[0].exception())

executor.shutdown(wait=True, cancel_futures=True)

from concurrent.futures import ThreadPoolExecutor
def normalize(number: int) -> str:
    return f"item-{number:06d}"
def huge_stream():
    for number in range(1_000_000):
        yield number
with ThreadPoolExecutor(max_workers=8) as executor:
    results = executor.map(normalize, huge_stream(), buffersize=32)
    for result in results:
        if result.endswith("000099"):
            print(result)
            break

# 危險概念：max_workers=1 時，外層工作占住唯一 worker，
# 又 submit 內層工作並呼叫 inner.result()，內層永遠沒機會開始。