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

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

一、并发编程基础

在开发高响应度的应用时，并发编程是绕不开的话题。简单来说，它允许程序同时执行多个任务，无论是线程还是进程，都能显著提升执行效率并充分利用系统资源。

1.1 核心优势与应用场景

并发编程的价值主要体现在三个方面：

• 提升效率：并行处理减少总运行时间。
• 资源利用：避免 CPU 或内存闲置。
• 结构优化：通过模块化设计简化复杂逻辑。

根据任务类型选择方案至关重要：CPU 密集型（如数学计算）适合多进程，而 I/O 密集型（如文件读写、网络请求）则更适合多线程。

二、多线程编程实践

2.1 线程的创建与管理

使用 threading 模块可以方便地管理线程。下面是一个简单的示例，展示如何启动并等待线程结束。

import threading
import time

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

# 创建线程
thread1 = threading.Thread(target=thread_function, args=('Thread 1',))
thread2 = threading.Thread(target=thread_function, args=('Thread 2',))

# 启动线程
thread1.start()
thread2.start()

# 等待线程结束
thread1.join()
thread2.join()
print('所有线程结束')

2.2 同步与互斥机制

当多个线程访问共享资源时，必须防止数据竞争。这里使用互斥锁（Lock）来保护计数器。

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}')

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 库，可以大幅降低异步开发的复杂度。建议在实际项目中多尝试这两种模式，根据性能瓶颈灵活调整。