完整解析 ThreadPoolExecutor：Python 高效併發工具

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    """ 模擬一個需要 2 秒的工作 """
    print(f"🔄 任務 {n} 開始執行")
    time.sleep(2)
    print(f"✅ 任務 {n} 完成")

# 建立一個執行緒池，最多允許 3 個執行緒同時運行
with ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(task, range(5))  # 執行 5 個任務

🔄 任務 0 開始執行
🔄 任務 1 開始執行
🔄 任務 2 開始執行
✅ 任務 0 完成
✅ 任務 1 完成
✅ 任務 2 完成
🔄 任務 3 開始執行
🔄 任務 4 開始執行
✅ 任務 3 完成
✅ 任務 4 完成

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor(max_workers=3) as executor:
    future1 = executor.submit(square, 2)
    future2 = executor.submit(square, 3)

    print(future1.result())  # 4
    print(future2.result())  # 9

def divide(n, d):
    if d == 0:
        raise ValueError("❌ 除數不能為 0")
    return n / d

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(divide, 10, i) for i in range(3)]  # 除以 0 會報錯

    for future in futures:
        try:
            print(future.result())  # 嘗試取得結果
        except Exception as e:
            print(f"⚠️ 錯誤: {e}")  # 捕捉異常

⚠️ 錯誤: ❌ 除數不能為 0
5.0
10.0 

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(square, [1, 2, 3, 4, 5])
    print(list(results))  # [1, 4, 9, 16, 25]

with ThreadPoolExecutor(max_workers=3) as executor:
    try:
        results = executor.map(divide, [10, 10, 10], [1, 0, 2])  # 除數有 0
        print(list(results))  # 這行不會執行，因為 `map()` 會在錯誤發生時中斷
    except Exception as e:
        print(f"⚠️ map() 發生錯誤: {e}")

⚠️ map() 發生錯誤: ❌ 除數不能為 0

from concurrent.futures import ThreadPoolExecutor
import requests

API_URLS = [
    "https://jsonplaceholder.typicode.com/todos/1",
    "https://jsonplaceholder.typicode.com/todos/2",
    "https://jsonplaceholder.typicode.com/todos/3"
]

def fetch_data(url):
    response = requests.get(url)
    return response.json()

# 使用 ThreadPoolExecutor 同時請求多個 API
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch_data, API_URLS))

print(results)  # 取得三個 API 回應的 JSON 數據

from concurrent.futures import ThreadPoolExecutor
import requests

URLS = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
]

def scrape_page(url):
    response = requests.get(url)
    return f"{url} 爬取完成，狀態碼: {response.status_code}"

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(scrape_page, URLS))

print("\n".join(results))

from concurrent.futures import ThreadPoolExecutor
import os

def read_file(file_path):
    with open(file_path, "r", encoding="utf-8") as file:
        return file.read()

files = ["file1.txt", "file2.txt", "file3.txt"]

with ThreadPoolExecutor(max_workers=3) as executor:
    contents = list(executor.map(read_file, files))

print(contents)  # 讀取多個文件的內容

from concurrent.futures import ProcessPoolExecutor

def heavy_computation(n):
    return sum(i**2 for i in range(n))

numbers = [1000000, 2000000, 3000000]

with ProcessPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(heavy_computation, numbers))

print(results)  # 處理三個大數據計算任務