线程池ThreadPoolExecutor详解 - Python并发编程指南

from concurrent.futures import ThreadPoolExecutor

# 创建固定大小的线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    # 提交任务
    future = executor.submit(task_func, arg1, arg2)
    # with块结束时自动关闭线程池

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=5)
try:
    future = executor.submit(task_func, arg1, arg2)
finally:
    executor.shutdown(wait=True)  # 等待所有任务完成

from concurrent.futures import ThreadPoolExecutor
import time

def task(name, duration):
    print(f"任务 {name} 开始执行")
    time.sleep(duration)
    print(f"任务 {name} 执行完成")
    return f"结果-{name}"

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交多个任务
    future1 = executor.submit(task, "A", 2)
    future2 = executor.submit(task, "B", 1)
    future3 = executor.submit(task, "C", 3)
    
    # 获取结果（阻塞等待）
    result1 = future1.result()
    result2 = future2.result()
    result3 = future3.result()
    
    print(f"所有结果: {result1}, {result2}, {result3}")

from concurrent.futures import ThreadPoolExecutor
import time

def square(n):
    time.sleep(0.5)
    return n * n

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

with ThreadPoolExecutor(max_workers=5) as executor:
    # 批量提交任务
    results = executor.map(square, numbers)
    
    # results是迭代器，按提交顺序返回结果
    for num, result in zip(numbers, results):
        print(f"{num}^2 = {result}")

from concurrent.futures import ThreadPoolExecutor
import time

def task(duration):
    time.sleep(duration)
    return f"完成-{duration}"

with ThreadPoolExecutor(max_workers=3) as executor:
    future = executor.submit(task, 2)
    
    # 检查任务状态
    print(f"是否运行中: {future.running()}")
    print(f"是否完成: {future.done()}")
    
    # 等待结果（可设置超时）
    result = future.result(timeout=5)
    print(f"结果: {result}")
    
    # 添加回调函数
    future2 = executor.submit(task, 1)
    future2.add_done_callback(lambda f: print(f"回调: {f.result()}"))

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import random

def download_file(url):
    """模拟文件下载"""
    print(f"开始下载: {url}")
    time.sleep(random.uniform(1, 3))
    print(f"下载完成: {url}")
    return f"下载成功-{url}"

urls = [
    "https://example.com/file1.zip",
    "https://example.com/file2.zip",
    "https://example.com/file3.zip",
    "https://example.com/file4.zip",
    "https://example.com/file5.zip",
]

# 使用线程池并发下载
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交所有任务
    future_to_url = {executor.submit(download_file, url): url for url in urls}
    
    # 使用as_completed按完成顺序处理结果
    for future in as_completed(future_to_url):
        url = future_to_url[future]
        try:
            result = future.result()
            print(f"{url}: {result}")
        except Exception as e:
            print(f"{url} 下载失败: {e}")

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_data(id):
    """获取数据"""
    time.sleep(0.5)
    return {"id": id, "value": id * 10}

def process_data(data):
    """处理数据"""
    time.sleep(0.3)
    data["processed"] = data["value"] * 2
    return data

def save_data(data):
    """保存数据"""
    time.sleep(0.2)
    print(f"保存数据: {data}")
    return True

# 数据处理管道
with ThreadPoolExecutor(max_workers=5) as executor:
    # 阶段1: 获取数据
    data_futures = [executor.submit(fetch_data, i) for i in range(10)]
    
    # 阶段2: 处理数据
    process_futures = []
    for future in data_futures:
        data = future.result()
        process_futures.append(executor.submit(process_data, data))
    
    # 阶段3: 保存数据
    save_futures = []
    for future in process_futures:
        data = future.result()
        save_futures.append(executor.submit(save_data, data))
    
    # 等待所有保存完成
    results = [f.result() for f in save_futures]
    print(f"所有数据保存完成: {sum(results)} 条")

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time

def slow_task(duration):
    time.sleep(duration)
    return f"任务完成-{duration}"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(slow_task, 1),
        executor.submit(slow_task, 2),
        executor.submit(slow_task, 5),  # 这个会超时
    ]
    
    for i, future in enumerate(futures):
        try:
            result = future.result(timeout=3)
            print(f"任务{i}: {result}")
        except TimeoutError:
            print(f"任务{i}: 执行超时")
        except Exception as e:
            print(f"任务{i}: 执行异常 - {e}")