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

2.3 线程池

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 进程的创建与管理

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 进程的通信

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

五、同步与互斥

4.1 锁的使用

import threading
import time

# 共享资源
counter = 0

# 互斥锁
lock = threading.Lock()

# 定义线程函数
def thread_function(name):
    global counter
    print(f'线程 {name} 开始')
    # 获取锁
    lock.acquire()
    try:
        # 修改共享资源
        counter += 1
        print(f'线程 {name} 已修改共享资源，值为 {counter}')
    finally:
        # 释放锁
        lock.release()
        time.sleep(2)
        print(f'线程 {name} 结束')

# 创建线程
threads = []
for i in range(5):
    thread = threading.Thread(target=thread_function, args=(f'Thread {i}',))
    threads.append(thread)
    thread.start()

# 等待所有线程结束
for thread in threads:
    thread.join()
print(f'所有线程结束，共享资源的值为 {counter}')

4.2 条件变量的使用

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)

5.3 实施过程

1. 安装 requests 库。
2. 定义下载文件的函数。
3. 定义并发下载文件的函数。
4. 测试程序。

5.4 最终效果

通过并发下载文件的程序，我们可以实现以下功能：

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

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

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)

6.3 实施过程

1. 安装 pandas 库。
2. 定义处理数据文件的函数。
3. 定义并发处理数据文件的函数。
4. 定义保存处理结果的函数。
5. 测试程序。

6.4 最终效果

通过并发数据处理的程序，我们可以实现以下功能：

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

八、总结

本文详细介绍了 Python 并发编程的基本概念和方法，包括多线程、多进程、线程池、进程池等；学习了 threading、multiprocessing 等核心库的使用；通过实战案例开发了并发下载文件和并发数据处理的程序。

建议读者在学习过程中多练习，通过编写代码加深对知识点的理解。