Python 并发编程实战：多线程、多进程与线程池

线程池管理

手动管理线程生命周期比较繁琐，concurrent.futures 中的 ThreadPoolExecutor 能自动处理线程复用和任务调度。

import concurrent.futures
import time

def thread_function(name):
    print(f'线程 {name} 开始')
    time.sleep(2)
    print(f'线程 {name} 结束')
    return name

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future1 = executor.submit(thread_function, 'Thread 1')
    future2 = executor.submit(thread_function, 'Thread 2')
    future3 = executor.submit(thread_function, 'Thread 3')

    print(f'线程 {future1.result()} 完成')
    print(f'线程 {future2.result()} 完成')
    print(f'线程 {future3.result()} 完成')
    print('所有线程结束')

多进程编程

对于 CPU 密集型任务，多进程可以绕过 GIL 限制，利用多核 CPU 的优势。multiprocessing 模块提供了类似线程的 API，但操作的是操作系统级别的进程。

进程的创建与管理

import multiprocessing
import time

def process_function(name):
    print(f'进程 {name} 开始')
    time.sleep(2)
    print(f'进程 {name} 结束')

process1 = multiprocessing.Process(target=process_function, args=('Process 1',))
process2 = multiprocessing.Process(target=process_function, args=('Process 2',))

process1.start()
process2.start()

process1.join()
process2.join()
print('所有进程结束')

进程通信

不同进程之间内存不共享，需要通过管道（Pipe）或队列（Queue）进行通信。

import multiprocessing
import time

def process_function(conn):
    print(f'子进程发送数据')
    conn.send('Hello from child process')
    time.sleep(2)
    print(f'子进程结束')
    conn.close()

parent_conn, child_conn = multiprocessing.Pipe()
process = multiprocessing.Process(target=process_function, args=(child_conn,))
process.start()

print(f'父进程接收数据：{parent_conn.recv()}')
process.join()
print('所有进程结束')

进程池

与线程池类似，ProcessPoolExecutor 用于管理进程池，适合批量处理独立任务。

import concurrent.futures
import time

def process_function(name):
    print(f'进程 {name} 开始')
    time.sleep(2)
    print(f'进程 {name} 结束')
    return name

with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor:
    future1 = executor.submit(process_function, 'Process 1')
    future2 = executor.submit(process_function, 'Process 2')
    future3 = executor.submit(process_function, 'Process 3')

    print(f'进程 {future1.result()} 完成')
    print(f'进程 {future2.result()} 完成')
    print(f'进程 {future3.result()} 完成')
    print('所有进程结束')

同步与互斥进阶

除了基本的锁，条件变量（Condition）常用于生产者 - 消费者模型，实现更复杂的线程协作。

import threading
import time

items = []
condition = threading.Condition()

def producer():
    for i in range(5):
        with condition:
            items.append(i)
            print(f'生产者生产了 {i}')
            condition.notify()
        time.sleep(1)

def consumer():
    for i in range(5):
        with condition:
            while not items:
                condition.wait()
            item = items.pop(0)
            print(f'消费者消费了 {item}')
        time.sleep(1)

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()
print('所有线程结束')

实战案例

并发下载文件

下载文件是典型的 I/O 密集型任务，使用线程池可以大幅提升速度。

import requests
import concurrent.futures
import os

def download_file(url, save_path):
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        with open(save_path, 'wb') as file:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    file.write(chunk)
        print(f'文件 {save_path} 下载成功')
        return save_path
    except Exception as e:
        print(f'文件 {save_path} 下载失败：{e}')
        return None

def download_files(urls, save_dir, max_workers=5):
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
    save_paths = [os.path.join(save_dir, os.path.basename(url)) for url in urls]
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(download_file, url, save_path) for url, save_path in zip(urls, save_paths)]
        
        for future in concurrent.futures.as_completed(futures):
            future.result()

if __name__ == '__main__':
    urls = [
        'https://www.example.com/page1.html',
        'https://www.example.com/page2.html',
        'https://www.example.com/page3.html',
        'https://www.example.com/page4.html',
        'https://www.example.com/page5.html'
    ]
    save_dir = 'downloads'
    download_files(urls, save_dir)

并发数据处理

如果是 CPU 密集型的统计计算，应切换到进程池以避免 GIL 瓶颈。

import pandas as pd
import concurrent.futures
import os

def process_file(file_path):
    try:
        df = pd.read_csv(file_path)
        stats = {
            '文件名': os.path.basename(file_path),
            '行数': df.shape[0],
            '列数': df.shape[1],
            '平均值': df.mean().to_dict(),
            '最大值': df.max().to_dict(),
            '最小值': df.min().to_dict()
        }
        print(f'文件 {os.path.basename(file_path)} 处理成功')
        return stats
    except Exception as e:
        print(f'文件 {os.path.basename(file_path)} 处理失败：{e}')
        return None

def process_files(file_paths, max_workers=5):
    results = []
    with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_file, file_path) for file_path in file_paths]
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result:
                results.append(result)
    return results

def save_results(results, save_path):
    df = pd.DataFrame(results)
    df.to_csv(save_path, index=False)
    print(f'处理结果已保存到 {save_path}')

if __name__ == '__main__':
    file_paths = ['data1.csv', 'data2.csv', 'data3.csv', 'data4.csv', 'data5.csv']
    save_path = 'results.csv'
    results = process_files(file_paths)
    save_results(results, save_path)

掌握并发编程不仅能优化现有代码的性能，还能让架构设计更加灵活。在实际项目中，根据任务特性选择合适的并发模型，配合合理的同步机制，是写出健壮高并发代码的关键。