Python 并发编程实战：多线程与多进程详解

2.3 线程池的使用

手动管理线程比较繁琐，concurrent.futures 提供的线程池能自动调度工作线程。

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('所有线程结束')

三、多进程编程实践

3.1 进程的创建与管理

对于 CPU 密集型任务，多进程能绕过 GIL 限制，真正利用多核 CPU。

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('所有进程结束')

3.2 进程间通信

进程之间默认隔离，需要通过管道（Pipe）或队列进行通信。

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('所有进程结束')

3.3 进程池的使用

类似线程池，进程池也支持批量提交任务。

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('所有线程结束')

五、实战案例：并发下载文件

在实际业务中，我们常需要批量下载资源。下面演示如何使用线程池加速这一过程。

5.1 需求分析

• 支持并发下载多个文件。
• 显示下载进度。
• 处理下载失败的情况。

5.2 代码实现

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)

六、实战案例：并发数据处理

面对大量 CSV 文件，单线程处理往往太慢。利用多进程可以显著缩短统计耗时。

6.1 需求分析

• 并发处理多个数据文件。
• 计算数据的统计信息。
• 保存处理结果。

6.2 代码实现

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):
    with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_file, file_path) for file_path in file_paths]
        results = []
        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)

七、结语

掌握并发编程的关键在于理解不同场景下的资源调度策略。I/O 操作优先选线程，计算密集优先选进程。配合 concurrent.futures 库，可以大幅降低异步开发的复杂度。建议在实际项目中多尝试这两种模式，根据性能瓶颈灵活调整。