Python concurrent.futures并发编程实战

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * n

with ThreadPoolExecutor(max_workers=4) as executor:
    future = executor.submit(task, 5)
    result = future.result()
    print(result)

from concurrent.futures import ThreadPoolExecutor

# 创建线程池
executor = ThreadPoolExecutor(
    max_workers=4,
    thread_name_prefix='worker-'
)

# 使用上下文管理器
with ThreadPoolExecutor(max_workers=4) as executor:
    # 提交任务
    pass

from concurrent.futures import ThreadPoolExecutor

def download_file(url):
    # 模拟下载
    import time
    time.sleep(1)
    return f"Downloaded: {url}"

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交单个任务
    future = executor.submit(download_file, "http://example.com/file1.txt")
    
    # 获取结果（阻塞）
    result = future.result(timeout=5)
    print(result)

urls = [
    "http://example.com/file1.txt",
    "http://example.com/file2.txt",
    "http://example.com/file3.txt",
]

with ThreadPoolExecutor(max_workers=3) as executor:
    # 批量提交
    futures = [executor.submit(download_file, url) for url in urls]
    
    # 获取所有结果
    for future in futures:
        print(future.result())

from concurrent.futures import ProcessPoolExecutor

def compute_heavy(n):
    # CPU密集型任务
    return sum(i * i for i in range(n))

with ProcessPoolExecutor(max_workers=4) as executor:
    future = executor.submit(compute_heavy, 1_000_000)
    print(future.result())

# IO密集型任务 → ThreadPoolExecutor
# CPU密集型任务 → ProcessPoolExecutor
# 混合任务 → 结合使用

def process_task(data):
    # IO操作
    raw_data = fetch_from_api(data)
    # CPU操作
    result = compute(raw_data)
    return result

from concurrent.futures import Future

future = Future()

# 检查状态
print(future.done())    # False
print(future.running()) # False
print(future.cancelled()) # False

# 设置结果
future.set_result(42)

print(future.done())    # True
print(future.result())  # 42

def callback(future):
    print(f"Task completed: {future.result()}")

with ThreadPoolExecutor() as executor:
    future = executor.submit(task, 5)
    future.add_done_callback(callback)

try:
    result = future.result(timeout=2)
except concurrent.futures.TimeoutError:
    print("Task timed out")

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(id):
    import time
    time.sleep(id)
    return f"Task {id} completed"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(1, 4)]
    
    # 按完成顺序获取结果
    for future in as_completed(futures):
        print(future.result())

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(task, [1, 2, 3, 4, 5])
    
    # 按输入顺序返回结果
    for result in results:
        print(result)

from concurrent.futures import wait, FIRST_COMPLETED

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(1, 4)]
    
    # 等待第一个完成
    done, not_done = wait(futures, return_when=FIRST_COMPLETED)
    
    print(f"Completed: {len(done)}")
    print(f"Not completed: {len(not_done)}")

import requests
from concurrent.futures import ThreadPoolExecutor

def download_file(url, save_path):
    response = requests.get(url)
    with open(save_path, 'wb') as f:
        f.write(response.content)
    return save_path

urls = [
    ("https://example.com/image1.jpg", "images/image1.jpg"),
    ("https://example.com/image2.jpg", "images/image2.jpg"),
    ("https://example.com/image3.jpg", "images/image3.jpg"),
]

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(download_file, url, path) for url, path in urls]
    
    for future in as_completed(futures):
        print(f"Downloaded: {future.result()}")

import psycopg2
from concurrent.futures import ThreadPoolExecutor

def query_user(user_id):
    conn = psycopg2.connect("dbname=example user=postgres")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    result = cursor.fetchone()
    conn.close()
    return result

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

with ThreadPoolExecutor(max_workers=4) as executor:
    results = executor.map(query_user, user_ids)
    
    for user_id, user in zip(user_ids, results):
        print(f"User {user_id}: {user}")

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def fetch_data(url):
    import requests
    return requests.get(url).json()

def process_data(data):
    # CPU密集型处理
    return sum(item['value'] for item in data)

def pipeline(url):
    data = fetch_data(url)
    return process_data(data)

urls = ["https://api.example.com/data1", "https://api.example.com/data2"]

# IO阶段使用线程池
with ThreadPoolExecutor(max_workers=4) as io_executor:
    futures = [io_executor.submit(fetch_data, url) for url in urls]
    raw_data = [f.result() for f in futures]

# CPU阶段使用进程池
with ProcessPoolExecutor(max_workers=4) as cpu_executor:
    results = list(cpu_executor.map(process_data, raw_data))

print(results)

import os

# CPU密集型任务
cpu_workers = os.cpu_count() or 4

# IO密集型任务
io_workers = min(32, (os.cpu_count() or 4) * 5)

# 不好的做法：共享可变状态
counter = 0

def increment():
    global counter
    counter += 1

# 好的做法：使用线程安全的数据结构或锁
from threading import Lock

class ThreadSafeCounter:
    def __init__(self):
        self._count = 0
        self._lock = Lock()
    
    def increment(self):
        with self._lock:
            self._count += 1

executor = ThreadPoolExecutor(max_workers=4)

try:
    # 提交任务
    futures = [executor.submit(task, i) for i in range(10)]
    
    # 获取结果
    for future in futures:
        print(future.result())
finally:
    # 关闭执行器
    executor.shutdown(wait=True)

import time

def sync_download(urls):
    for url in urls:
        download_file(url)

def async_download(urls):
    with ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_file, urls)

# 性能对比
urls = ["https://example.com/file{}.txt".format(i) for i in range(10)]

start = time.time()
sync_download(urls)
print(f"Sync time: {time.time() - start:.2f}s")

start = time.time()
async_download(urls)
print(f"Async time: {time.time() - start:.2f}s")

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

# 线程池（受GIL限制）
with ThreadPoolExecutor(max_workers=4) as executor:
    start = time.time()
    executor.map(cpu_intensive, [10_000_000] * 4)
    print(f"ThreadPool time: {time.time() - start:.2f}s")

# 进程池（不受GIL限制）
with ProcessPoolExecutor(max_workers=4) as executor:
    start = time.time()
    executor.map(cpu_intensive, [10_000_000] * 4)
    print(f"ProcessPool time: {time.time() - start:.2f}s")